前言

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

谨以此文，献给远方的老铁

一、目标

二、使用步骤

1. 安装 consul

我们可以直接使用官方提供的二进制文件来进行安装部署，其官网地址为 https://www.consul.io/downloads
image.png
下载后为可执行文件，在我们开发试验过程中，可以直接使用 consul agent -dev 命令来启动一个单节点的 consul

在启动的打印日志中可以看到 agent: Started HTTP server on 127.0.0.1:8500 (tcp), 我们可以在浏览器直接访问 127.0.0.1:8500 即可看到如下
image.png
这里我们的 consul 就启动成功了

2. 服务注册

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

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

  1. class ServiceInstance:
  3.     def __init__(self, service_id: str, host: str, port: int, secure: bool = False, metadata: dict = None,
  4.                  instance_id: str = None):
  5.         self.service_id = service_id
  6.         self.host = host
  7.         self.port = port
  8.         self.secure = secure
  9.         self.metadata = metadata
  10.         self.instance_id = instance_id
  12.     def get_instance_id(self):
  13.         return

定义基类

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

  1. import abc
  4. class ServiceRegistry(abc.ABC):
  6.     @abc.abstractmethod
  7.     def register(self, service_instance: ServiceInstance):
  8.         pass
  10.     @abc.abstractmethod
  11.     def deregister(self):
  12.         pass

具体实现

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

  1. import consul
  4. class ConsulServiceRegistry(ServiceRegistry):
  5.     _consul = None
  6.     _instance_id = None
  8.     def __init__(self, host: str, port: int, token: str = None):
  9.         self.host = host
  10.         self.port = port
  11.         self.token = token
  12.         self._consul = consul.Consul(host, port, token=token)
  14.     def register(self, service_instance: ServiceInstance):
  15.         schema = "http"
  16.         if service_instance.secure:
  17.             schema = "https"
  18.         check = consul.Check.http(f'{schema}:{service_instance.host}:{service_instance.port}/actuator/health', "1s",
  19.                                   "3s", "10s")
  20.         self._consul.agent.service.register(service_instance.service_id,
  21.                                             service_id=service_instance.instance_id,
  22.                                             address=service_instance.host,
  23.                                             port=service_instance.port,
  24.                                             check=check)
  25.         self._instance_id = service_instance.instance_id
  27.     def deregister(self):
  28.         if self._instance_id:
  29.             self._consul.agent.service.deregister(service_id=self._instance_id)
  30.             self._instance_id = None

3. 服务发现

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

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

基类定义

  1. import abc
  4. class DiscoveryClient(abc.ABC):
  6.     @abc.abstractmethod
  7.     def get_services(self) -> list:
  8.         pass
  10.     @abc.abstractmethod
  11.     def get_instances(self, service_id: str) -> list:
  12.         pass

具体实现

来实现一下

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

  1. import consul
  4. class ConsulServiceDiscovery(DiscoveryClient):
  6.     _consul = None
  8.     def __init__(self, host: str, port: int, token: str = None):
  9.         self.host = host
  10.         self.port = port
  11.         self.token = token
  12.         self._consul = consul.Consul(host, port, token=token)
  14.     def get_services(self) -> list:
  15.         return self._consul.catalog.services()[1].keys()
  17.     def get_instances(self, service_id: str) -> list:
  18.         origin_instances = self._consul.catalog.service(service_id)[1]
  19.         result = []
  20.         for oi in origin_instances:
  21.             result.append(ServiceInstance(
  22.                 oi.get('ServiceName'),
  23.                 oi.get('ServiceAddress'),
  24.                 oi.get('ServicePort'),
  25.                 oi.get('ServiceTags'),
  26.                 oi.get('ServiceMeta'),
  27.                 oi.get('ServiceID'),
  28.             ))
  29.         return result

4. 测试用例

  1. import unittest
  2. from random import random
  5. class MyTestCase(unittest.TestCase):
  6.     def test_consul_register(self):
  7.         instance = ServiceInstance("abc", "127.0.0.1", 8000, instance_id=f'abc_{random()}')
  9.         registry = ConsulServiceRegistry("127.0.0.1", 8500)
  10.         discovery = ConsulServiceDiscovery("127.0.0.1", 8500)
  11.         registry.register(instance)
  12.         print(discovery.get_services())
  13.         print(discovery.get_instances("abc"))
  14.         self.assertEqual(True, True)
  17. if __name__ == '__main__':
  18.     unittest.main()

总结

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

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

参考