1.使用request解决

  1. import requests
  3. headers = {
  4.     "contentType": "application/json"
  5. }
  8. def register(name, id, address, port):
  9.     url = "http://192.168.119.128:8500/v1/agent/service/register"
  10.     rsp = requests.put(url, headers=headers, json={
  11.         "Name": name,
  12.         "ID": id,
  13.         "Tags": ["mxshop", "bobby", "imooc", "web"],
  14.         "Address": address,
  15.         "Port": port,
  17.     })
  18.     if rsp.status_code == 200:
  19.         print("注册成功")
  20.     else:
  21.         print(f"注册失败: {rsp.status_code}")
  24. def deregistry(id):
  25.     url = f"http://192.168.119.128:8500/v1/agent/service/deregister/{id}"
  26.     rsp = requests.put(url, headers=headers)
  27.     if rsp.status_code == 200:
  28.         print("注销成功")
  29.     else:
  30.         print(f"注销失败: {rsp.status_code}")
  33. if __name__ == "__main__":
  34.     # register("mxshop-web", "mxshop-web", "127.0.0.1", 50051)
  35.     deregistry("mxshop-web")

2.健康检查的服务注册

  1. import requests
  3. headers = {
  4.     "contentType": "application/json"
  5. }
  8. def register(name, id, address, port):
  9.     url = "http://192.168.119.128:8500/v1/agent/service/register"
  10.     rsp = requests.put(url, headers=headers, json={
  11.         "Name": name,
  12.         "ID": id,
  13.         "Tags": ["mxshop", "bobby", "imooc", "web"],
  14.         "Address": address,
  15.         "Port": port,
  16.         "Check": {
  17.             "HTTP": f"http://{address}:{port}/health",
  18.             "Timeout": "5s",
  19.             "Interval": "5s",
  20.             "DeregisterCriticalServiceAfter": "15s",
  21.         }
  23.     })
  24.     if rsp.status_code == 200:
  25.         print("注册成功")
  26.     else:
  27.         print(f"注册失败: {rsp.status_code}")
  30. def deregistry(id):
  31.     url = f"http://192.168.119.128:8500/v1/agent/service/deregister/{id}"
  32.     rsp = requests.put(url, headers=headers)
  33.     if rsp.status_code == 200:
  34.         print("注销成功")
  35.     else:
  36.         print(f"注销失败: {rsp.status_code}")
  39. if __name__ == "__main__":
  40.     register("mxshop-web", "mxshop-web", "192.168.245.1", 8021)
  41.     # deregistry("mxshop-web")

3.过滤服务

  1. def filter_service(name):
  2.     url = "http://192.168.119.128:8500/v1/agent/services"
  3.     params = {
  4.         "filter": f'Service == "{name}"'
  5.     }
  6.     rsp = requests.get(url, params=params).json()
  7.     for key, value in rsp.items():
  8.         print(key)

4.使用第三方库

安装

  1. pip install python-consul2
  1. import consul
  2. import json
  3. c = consul.Consul(host="192.168.119.128")
  4. address = "192.168.119.129"
  5. port = 50051
  6. check = {
  7.     "GRPC": f"{address}:{port}",
  8.     "GRPCUseTLS": False,  # 不需要做安全验证
  9.     "Timeout": "5s",
  10.     "Interval": "5s",
  11.     "DeregisterCriticalServiceAfter": "15s",
  12. }
  13. rsp = c.agent.service.register(name="user-srv", service_id="user-srv", address=address, port=port,
  14.                                tags=["mxshop"], check=check)
  15. rsp = c.agent.services()
  16. json.loads(rsp)
  17. rsp = c.agent.service.deregister("user-srv")
  18. print(rsp)