Python 使用 consul 做服務(wù)發(fā)現(xiàn)示例詳解

更新時(shí)間：2021年03月10日 10:09:56 作者：雙鬼帶單

這篇文章主要介紹了Python 使用 consul 做服務(wù)發(fā)現(xiàn)示例詳解,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

前言

前面一章講了微服務(wù)的一些優(yōu)點(diǎn)和缺點(diǎn)，那如何做到

一、目標(biāo)

二、使用步驟

1. 安裝 consul

我們可以直接使用官方提供的二進(jìn)制文件來(lái)進(jìn)行安裝部署，其官網(wǎng)地址為 https://www.consul.io/downloads

在這里插入圖片描述

下載后為可執(zhí)行文件，在我們開(kāi)發(fā)試驗(yàn)過(guò)程中，可以直接使用 consul agent -dev 命令來(lái)啟動(dòng)一個(gè)單節(jié)點(diǎn)的 consul

在啟動(dòng)的打印日志中可以看到 agent: Started HTTP server on 127.0.0.1:8500 (tcp), 我們可以在瀏覽器直接訪問(wèn) 127.0.0.1:8500 即可看到如下

在這里插入圖片描述

這里我們的 consul 就啟動(dòng)成功了

2. 服務(wù)注冊(cè)

在網(wǎng)絡(luò)編程中，一般會(huì)提供項(xiàng)目的 IP、PORT、PROTOCOL，在服務(wù)治理中，我們還需要知道對(duì)應(yīng)的服務(wù)名、實(shí)例名以及一些自定義的擴(kuò)展信息

在這里使用 ServiceInstance 接口來(lái)規(guī)定注冊(cè)服務(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ī)定了需要注冊(cè)的服務(wù)的必要信息，下面定義下服務(wù)注冊(cè)和剔除的方法，方便以后實(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 接口來(lái)對(duì)consul 進(jìn)行操作，我們也可以使用 http 請(qǐng)求方式進(jìn)行注冊(cè)和剔除操作，具體 http 接口文檔見(jiàn) https://www.consul.io/api-docs, consul 并沒(méi)有提供 Python 語(yǔ)言的實(shí)現(xiàn)，這里使用 python-consul 來(lái)訪問(wèn) 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)中，一般會(huì)需要兩個(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)

來(lái)實(shí)現(xiàn)一下

這里是簡(jiǎ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. 測(cè)試用例

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é)

通過(guò)使用 consul api 我們可以簡(jiǎn)單的實(shí)現(xiàn)基于 consul 的服務(wù)發(fā)現(xiàn)，在通過(guò)結(jié)合 http rpc 就可簡(jiǎn)單的實(shí)現(xiàn)服務(wù)的調(diào)用，下面一章來(lái)簡(jiǎn)單講下 go 如何發(fā)起 http 請(qǐng)求，為我們做 rpc 做個(gè)鋪墊

具體代碼見(jiàn) 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)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: