review_experts/expert_server.py

# coding:utf-8
'''
微服务服务端
'''
from proto import exists_expert_pb2
from servicerd.service import ServiceManage
from proto import exists_expert_pb2_grpc
from concurrent import futures
from loguru import logger
from predict import extract_expert
import grpc
import argparse
import time
import json

logger.add('./logs/runtime_{time}.log', rotation='00:00')

parser = argparse.ArgumentParser()
parser.add_argument('-host', '--serve_host', type=str, default="192.168.20.69", help='提供服务的主机地址')
parser.add_argument('-port', '--serve_port', type=int, default=14001, help='提供>服务的主机端口')
parser.add_argument('-chost', '--consul_serve_host', type=str, default='192.168.3.12', help='consul主机地址')
parser.add_argument('-cport', '--consul_serve_port', type=int, default=10021, help='consul主机端口')

args = parser.parse_args()

_ONE_DAY_IN_SECONDS = 60 * 60 * 24
_SERVICE_NAME = 'extract_expert_service'
_HOST = args.serve_host
_PORT = args.serve_port
RD_SERVER = "{}:{}".format(args.consul_serve_host, args.consul_serve_port)
WORKERS = 10


class ExtractExpertServices(exists_expert_pb2_grpc.ExistsExpertServicer):

    def extract(self, request, context):
        # TODO 调业务
        try:
            request_params = json.loads(request.contents)
            print(request_params)
            predict_result = extract_expert(request_params)
            result = json.dumps({"result": predict_result, "status": 200})
            return exists_expert_pb2.ExpertResponse(results=result)
        except Exception as e:
            logger.warning(e)
            result = json.dumps({"status": 300})
            return exists_expert_pb2.ExpertResponse(results=result)


def start_server(destory_fn: any):
    """
    启动服务
    :param destory_fn:
    :return:
    """
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=WORKERS), options=[
        ('grpc.max_send_message_length', 100 * 1024 * 1024),
        ('grpc.max_receive_message_length', 100 * 1024 * 1024)])
    exists_expert_pb2_grpc.add_ExistsExpertServicer_to_server(ExtractExpertServices(), server)
    server.add_insecure_port('[::]:%d' % (_PORT))
    server.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        destory_fn()
        server.stop(0)


if __name__ == '__main__':
    sm = ServiceManage(rd_server=RD_SERVER,
                       name=_SERVICE_NAME,
                       local_ip=_HOST,
                       local_port=_PORT,
                       workers=WORKERS,
                       balance_type=2)
    start_server(sm.destory)