Springboot构建的微服务，如果想用Consul来进行服务注册、发现等，已经有成熟的starter封装了相关逻辑，可以直接使用。
最近有粉丝咨询，Python的Web服务如何注册到Consul上？
今天工作之余，就写一下吧。
服务实例首先我们先定义服务实例:
class ServiceInstance: def __init__(self, service_id:str, host: str, port: int, tags: list = None, metadata: dict = None, instance_id: str = None): self.service_id = service_id self.host = host self.port = port self.tags = tags self.metadata = metadata self.instance_id = instance_id def get_instance_id(self): return服务注册
接着我们定义下注册服务的接口:
import abc class ServiceRegistry(abc.ABC): @abc.abstractmethod def register(self, service_instance: ServiceInstance): pass @abc.abstractmethod def degister(self): pass
写一下上述接口的Consul实现:
# consul包封装了Consul组件的相关操作，使用起来方便一些
# 当然你也可以自己构建Http请求来进行服务的注册和发现
# 但秉着不重复造轮子的态度，建议直接使用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"
# 定义服务状态的健康检查接口
check = consul.Check.http(f'{schema}://{service_instance.host}:{service_instance.port}/actuator/health', "10s", "10s", "10s")
// 调用consul的接口来实现服务注册
self._consul.agent.service.register(service_id = service_instance.service_id,
address = service_instance.host,
port = service_instance.port,
tags = service_instance.tags,
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
所谓服务状态的健康检查接口，指的是Consul会定周期的请求服务的某个接口，如果该接口有正常返回，证明该服务还活着。
如果不能正常返回，代表该服务下线了，Consul会将该服务标记为不可用，并在一定时间后将其从服务列表中清除。
所以想将服务注册到Consul上的话，需要额外实现1个健康检查接口，并将健康检查接口的地址在注册的时候，以请求参数的方式告知Consul，这样Consul才能监测该服务的状态。
如我们的Python服务是基于flask框架构建的，则健康检查接口可以定义为:
from flask import Flask
import json
app = Flask(__name__)
@app.route('/actuator/health')
def service_health():
status = {"status": "UP"}
return json.dumps(status)
启动服务后，再执行一下注册逻辑，就可以发现我们的Python服务成功注册到Consul上去了。
服务发现聊完服务注册，接着实现一下服务发现。
老规矩，先定义服务发现的接口类:
import abc class DiscoveryClient(abc.ABC): # 获取注册中心上的服务列表 @abc.abstractment def get_services(self) -> list: pass # 获取某个服务的实例列表 @abc.abstractment def get_instances(self, service_id: str) -> list: pass
接着写上述接口类的Consul实现:
import consul
class ConsulDiscoveryClient(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:
origin_instances_keys = self._consul.catalog.services()[1].keys()
result = []
for origin_instance_key in origin_instance_keys:
result.append(origin_instance_key)
return result
def get_instances(self, service_id: str) -> list:
origin_instances = self._consul.catalog.service(service_id)[1]
result = []
for origin_instance in origin_instances:
result.append(ServiceInstance(
origin_instance.get('ServiceName'),
origin_instance.get('ServiceAddress'),
origin_instance.get('ServicePort'),
origin_instance.get('ServiceTags'),
origin_instance.get('Servicemeta'),
origin_instance.get('ServiceID'),
))
return result
这样，我们就把Python服务注册到Consul上的功能实现了。
照猫画虎，你也可以自行实现如何将Python的Web服务注册到Nacos等其他注册中心。
当然，上面的代码没有考虑线程安全、Https协议等复杂的情形。
读者可自行完善上述代码，使其更加健壮。
