个人中心
gpt4 book ai didi

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

转载 作者：qq735679552 更新时间：2022-09-29 22:32:09 26 4
gpt4 key购买 nike

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章Python 使用 consul 做服务发现示例详解由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

前言

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

1、目标

2、使用步骤

1. 安装 consul

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

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

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

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

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

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

2. 服务注册

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

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

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

定义基类 。

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

import abcclass ServiceRegistry(abc.ABC): @abc.abstractmethod def register(self, service_instance: ServiceInstance): pass @abc.abstractmethod def deregister(self): pass

具体实现 。

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

import consulclass 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. 服务发现

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

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

基类定义 。

import abcclass DiscoveryClient(abc.ABC): @abc.abstractmethod def get_services(self) -> list: pass @abc.abstractmethod def get_instances(self, service_id: str) -> list: pass

具体实现 。

来实现一下 。

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

import consulclass 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 unittestfrom random import randomclass 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()

总结

通过使用 consul api 我们可以简单的实现基于 consul 的服务发现，在通过结合 http rpc 就可简单的实现服务的调用，下面一章来简单讲下 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 服务内容请搜索我以前的文章或继续浏览下面的相关文章希望大家以后多多支持我！ 。

原文链接：https://blog.csdn.net/zyndev/article/details/113888620 。

最后此篇关于Python 使用 consul 做服务发现示例详解的文章就讲到这里了,如果你想了解更多关于Python 使用 consul 做服务发现示例详解的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。

26 4 0
文章推荐： Python 平方列表中每个数字的多种操作
文章推荐： 超级好用的4个Python命令行可视化库
文章推荐： 基于Python+Pyqt5开发一个应用程序
文章推荐： python变量作用域与列表入门详解
qq735679552
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com