目錄
- 前言
- 一、目標(biāo)
- 二、使用步驟
- 1. 安裝 consul
- 2. 服務(wù)注冊
- 3. 服務(wù)發(fā)現(xiàn)
- 4. 測試用例
- 總結(jié)
前言
前面一章講了微服務(wù)的一些優(yōu)點(diǎn)和缺點(diǎn)，那如何做到
一、目標(biāo)
二、使用步驟
1. 安裝 consul
我們可以直接使用官方提供的二進(jìn)制文件來進(jìn)行安裝部署，其官網(wǎng)地址為 https://www.consul.io/downloads
下載后為可執(zhí)行文件，在我們開發(fā)試驗(yàn)過程中，可以直接使用 consul agent -dev 命令來啟動一個(gè)單節(jié)點(diǎn)的 consul
在啟動的打印日志中可以看到 agent: Started HTTP server on 127.0.0.1:8500 (tcp), 我們可以在瀏覽器直接訪問 127.0.0.1:8500 即可看到如下
這里我們的 consul 就啟動成功了
2. 服務(wù)注冊
在網(wǎng)絡(luò)編程中，一般會提供項(xiàng)目的 IP、PORT、PROTOCOL，在服務(wù)治理中，我們還需要知道對應(yīng)的服務(wù)名、實(shí)例名以及一些自定義的擴(kuò)展信息
在這里使用 ServiceInstance 接口來規(guī)定注冊服務(wù)時(shí)必須的一些信息
class ServiceInstance:
def __init__(self, service_id: str, host: str, port: int, secure: bool = False, metadata: dict = None,
instance_id: str = None):
self.service_id = service_id
self.host = host
self.port = port
self.secure = secure
self.metadata = metadata
self.instance_id = instance_id
def get_instance_id(self):
return
定義基類
在上面規(guī)定了需要注冊的服務(wù)的必要信息，下面定義下服務(wù)注冊和剔除的方法，方便以后實(shí)現(xiàn) Eureka 和 Redis 的方式
import abc
class ServiceRegistry(abc.ABC):
@abc.abstractmethod
def register(self, service_instance: ServiceInstance):
pass
@abc.abstractmethod
def deregister(self):
pass
具體實(shí)現(xiàn)
因?yàn)?consul 提供了 http 接口來對consul 進(jìn)行操作，我們也可以使用 http 請求方式進(jìn)行注冊和剔除操作，具體 http 接口文檔見 https://www.consul.io/api-docs, consul 并沒有提供 Python 語言的實(shí)現(xiàn)，這里使用 python-consul 來訪問 consul
import consul
class ConsulServiceRegistry(ServiceRegistry):
_consul = None
_instance_id = None
def __init__(self, host: str, port: int, token: str = None):
self.host = host
self.port = port
self.token = token
self._consul = consul.Consul(host, port, token=token)
def register(self, service_instance: ServiceInstance):
schema = "http"
if service_instance.secure:
schema = "https"
check = consul.Check.http(f'{schema}:{service_instance.host}:{service_instance.port}/actuator/health', "1s",
"3s", "10s")
self._consul.agent.service.register(service_instance.service_id,
service_id=service_instance.instance_id,
address=service_instance.host,
port=service_instance.port,
check=check)
self._instance_id = service_instance.instance_id
def deregister(self):
if self._instance_id:
self._consul.agent.service.deregister(service_id=self._instance_id)
self._instance_id = None
3. 服務(wù)發(fā)現(xiàn)
在服務(wù)發(fā)現(xiàn)中，一般會需要兩個(gè)方法
- 獲取所有的服務(wù)列表
- 獲取指定的服務(wù)的所有實(shí)例信息
基類定義
import abc
class DiscoveryClient(abc.ABC):
@abc.abstractmethod
def get_services(self) -> list:
pass
@abc.abstractmethod
def get_instances(self, service_id: str) -> list:
pass
具體實(shí)現(xiàn)
來實(shí)現(xiàn)一下
這里是簡化版，所以一些參數(shù)直接寫死了，如果需要可以適當(dāng)修改
import consul
class ConsulServiceDiscovery(DiscoveryClient):
_consul = None
def __init__(self, host: str, port: int, token: str = None):
self.host = host
self.port = port
self.token = token
self._consul = consul.Consul(host, port, token=token)
def get_services(self) -> list:
return self._consul.catalog.services()[1].keys()
def get_instances(self, service_id: str) -> list:
origin_instances = self._consul.catalog.service(service_id)[1]
result = []
for oi in origin_instances:
result.append(ServiceInstance(
oi.get('ServiceName'),
oi.get('ServiceAddress'),
oi.get('ServicePort'),
oi.get('ServiceTags'),
oi.get('ServiceMeta'),
oi.get('ServiceID'),
))
return result
4. 測試用例
import unittest
from random import random
class MyTestCase(unittest.TestCase):
def test_consul_register(self):
instance = ServiceInstance("abc", "127.0.0.1", 8000, instance_id=f'abc_{random()}')
registry = ConsulServiceRegistry("127.0.0.1", 8500)
discovery = ConsulServiceDiscovery("127.0.0.1", 8500)
registry.register(instance)
print(discovery.get_services())
print(discovery.get_instances("abc"))
self.assertEqual(True, True)
if __name__ == '__main__':
unittest.main()
總結(jié)
通過使用 consul api 我們可以簡單的實(shí)現(xiàn)基于 consul 的服務(wù)發(fā)現(xiàn)，在通過結(jié)合 http rpc 就可簡單的實(shí)現(xiàn)服務(wù)的調(diào)用，下面一章來簡單講下 go 如何發(fā)起 http 請求，為我們做 rpc 做個(gè)鋪墊
具體代碼見 https://github.com/zhangyunan1994/gimini
參考
https://www.consul.io/api-docs
https://github.com/hashicorp/consul/tree/master/api
到此這篇關(guān)于Python 使用 consul 做服務(wù)發(fā)現(xiàn)的文章就介紹到這了,更多相關(guān)Python 使用 consul 服務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:- python實(shí)現(xiàn)FTP文件傳輸?shù)姆椒ǎǚ?wù)器端和客戶端）
- python實(shí)現(xiàn)從ftp服務(wù)器下載文件
- Python3搭建http服務(wù)器的實(shí)現(xiàn)代碼
- python+selenium定時(shí)爬取丁香園的新型冠狀病毒數(shù)據(jù)并制作出類似的地圖（部署到云服務(wù)器）
