consul在项目中使用

1.目录存放

创建公共目录存放common/register，创建base.py register.py

 2.base.py基础类

import abc
class Resister(metaclass=abc.ABCMeta):
@abc.abstractmethod
def register(self, name, id, address, port):
pass
@abc.abstractmethod
def deregister(self, service_id):
pass
@abc.abstractmethod
def get_all_service(self):
pass
@abc.abstractmethod
def filter_service(self, f):
pass

　3. 注册文件consul.py

import requests
from common.register import base
import consul
class ConsulRegister(base.Resister):
def __int__(self, host, port):
self.host = host
self.port = port
self.c = consul.Consul(host=host, port=port)
def register(self, name, id, address, port):
check = {
"GRPC": f"{address}:{port}",
"GRPCUseTLS": False,
"Timeout": "5s",
"Interval": "5s",
"DeregisterCriticalServiceAfter": "15s"
}
# 注册
return self.c.agent.service.register(name="user-srv", service_id="user-srv2",
address=address, port=port, tags=["mxshop"], check=check)
def deregister(self, service_id):
return self.c.agent.service.deregister(service_id)
def get_all_service(self):
return self.c.agent.services()
def filter_service(self, f):
url = f"http://{self.host}:{self.port}/v1/agent/services"
params = {
"filter": f
}
rsp = requests.get(url, params=params).json()
return rsp

　4. 使用

在server.py中使用

logger.info(f"服务注册开始")
register = consul.ConsulRegister(settings.CONSUL_HOST, settings.CONSUL_POST)
if not register.register(name=settings.SERVICE_NAME, id=settings.SERVICE_NAME, address=args.ip, port=args.port,
tags=settings.SERVICE_TAGS):
logger.info(f"服务注册失败")
sys.exit(0)
logger.info(f"服务注册成功")