Python 使用 consul 做服务发现示例详解

2023-07-04 11:30:03 317

前言

前面一章讲了微服务的一些优点和缺点，那如何做到

一、目标

二、使用步骤

1.安装consul

我们可以直接使用官方提供的二进制文件来进行安装部署，其官网地址为https://www.consul.io/downloads

下载后为可执行文件，在我们开发试验过程中，可以直接使用consulagent-dev命令来启动一个单节点的consul

在启动的打印日志中可以看到agent:StartedHTTPserveron127.0.0.1:8500(tcp),我们可以在浏览器直接访问127.0.0.1:8500即可看到如下

这里我们的consul就启动成功了

2.服务注册

在网络编程中，一般会提供项目的IP、PORT、PROTOCOL，在服务治理中，我们还需要知道对应的服务名、实例名以及一些自定义的扩展信息

在这里使用ServiceInstance接口来规定注册服务时必须的一些信息

classServiceInstance:

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

defget_instance_id(self):
return

定义基类

在上面规定了需要注册的服务的必要信息，下面定义下服务注册和剔除的方法，方便以后实现Eureka和Redis的方式

importabc


classServiceRegistry(abc.ABC):

@abc.abstractmethod
defregister(self,service_instance:ServiceInstance):
pass

@abc.abstractmethod
defderegister(self):
pass

具体实现

因为consul提供了http接口来对consul进行操作，我们也可以使用http请求方式进行注册和剔除操作，具体http接口文档见https://www.consul.io/api-docs,consul并没有提供Python语言的实现，这里使用python-consul来访问consul

importconsul


classConsulServiceRegistry(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)

defregister(self,service_instance:ServiceInstance):
schema="http"
ifservice_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

defderegister(self):
ifself._instance_id:
self._consul.agent.service.deregister(service_id=self._instance_id)
self._instance_id=None

3.服务发现

在服务发现中，一般会需要两个方法

获取所有的服务列表
获取指定的服务的所有实例信息

基类定义

importabc


classDiscoveryClient(abc.ABC):

@abc.abstractmethod
defget_services(self)->list:
pass

@abc.abstractmethod
defget_instances(self,service_id:str)->list:
pass

具体实现

来实现一下

这里是简化版，所以一些参数直接写死了，如果需要可以适当修改

importconsul


classConsulServiceDiscovery(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)

defget_services(self)->list:
returnself._consul.catalog.services()[1].keys()

defget_instances(self,service_id:str)->list:
origin_instances=self._consul.catalog.service(service_id)[1]
result=[]
foroiinorigin_instances:
result.append(ServiceInstance(
oi.get('ServiceName'),
oi.get('ServiceAddress'),
oi.get('ServicePort'),
oi.get('ServiceTags'),
oi.get('ServiceMeta'),
oi.get('ServiceID'),
))
returnresult

4.测试用例

importunittest
fromrandomimportrandom

classMyTestCase(unittest.TestCase):
deftest_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()

总结

通过使用consulapi我们可以简单的实现基于consul的服务发现，在通过结合httprpc就可简单的实现服务的调用，下面一章来简单讲下go如何发起http请求，为我们做rpc做个铺垫

具体代码见https://github.com/zhangyunan1994/gimini

参考

https://www.consul.io/api-docs

https://github.com/hashicorp/consul/tree/master/api

到此这篇关于Python使用consul做服务发现的文章就介绍到这了,更多相关Python使用consul服务内容请搜索毛票票以前的文章或继续浏览下面的相关文章希望大家以后多多支持毛票票！

声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。