python與consul 實現gRPC服務注冊-發現

背景

通過對gRPC的介紹我們知道，當正常啟動服務后，我們只需要知道ip，port就可以進行gRPC的連接。可以想到，這種方式並不適合用於線上環境，因為這樣直連的話就失去了擴展性，當需要多機部署的時候，就無法在線上環境直接使用，而且當線上項目連接的那台服務器宕了的話，整個項目也會出錯，這並不是我們想要的結果。
於是,我們需要一個服務注冊與發現的機制。也就是說當我們的rpc服務啟動的時候注冊到另一個服務器，然后客戶端連接的時候去查找對應的服務，得到相應的ip，port，然后就可以順利進行連接了。這種方式也就是服務注冊與發現，目前有zoomkeper, consul, 因為自己對zookeeper不熟悉，所以這里選用consul。整個流程如圖所示

consul 的安裝

1. 通過docker
   運行如下命令:
   
   docker run -d -p 8500:8500 consul consul agent -data-dir=/consul/data -config-dir=/consul/config -dev -client=0.0.0.0 -bind=0.0.0.0

2. 通過其它方式
   安裝方式參考https://www.consul.io/intro/getting-started/install.html
   安裝完后運行：
   consul agent --data-dir . -server -ui -bootstrap -bind=127.0.0.1

無論哪種方式，運行完之后在瀏覽器中打開 http://127.0.0.1:8500/ui， 可以看到如下內容

服務注冊

已之前介紹的gRPC代碼為基礎，我們加入服務注冊部分（注：本人環境為python3， 需要python2的，自己進行修改）

 import time
 import grpc
 import consul
 import json
 from concurrent import futures
 
 import test_pb2_grpc
 import test_pb2
 
 
 def test(request, context):
     # 實際調用到的函數
     json_response = test_pb2.JSONResponse()
     json_response.rst_string = json.dumps({"ret":"Hi gRPC"})# 構造出proto文件中定義的返回值格式
     return json_response
 
 class OrderHandler(test_pb2_grpc.OrderHandlerServicer):
     def create_order(self, request, context):
         return test(request, context)
 
 def register(server_name, ip, port):
     c = consul.Consul() # 連接consul 服務器，默認是127.0.0.1，可用host參數指定host
     print(f"開始注冊服務{server_name}")
     check = consul.Check.tcp(ip, port, "10s") # 健康檢查的ip，端口，檢查時間
     c.agent.service.register(server_name, f"{server_name}-{ip}-{port}",
             address=ip, port=port, check=check) # 注冊服務部分
     print(f"注冊服務{server_name}成功")
 
 def unregister(server_name, ip, port):
     c = consul.Consul()
     print(f"開始退出服務{server_name}")
     c.agent.service.deregister(f'{server_name}-{ip}-{port}')
 
 def serve():
     server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
     test_pb2_grpc.add_OrderHandlerServicer_to_server(OrderHandler(), server)
     server.add_insecure_port('[::]:{}'.format(12006))
     register("order_server", "0.0.0.0", 12006)
     server.start()
     try:
         while True:
             time.sleep(186400)
     except KeyboardInterrupt:
         unregister("order_server", "0.0.0.0", 12006)
         server.stop(0)
 
 serve()

要運行此服務需要先安裝 python-consul pip install python-consul
運行此服務，在瀏覽器中會出現我們剛剛注冊的服務，和可用節點，健康信息，如圖

當使用ctrl+c 退出服務時，會取消注冊。也就是會從consul列表里消失。
注意： 當健康檢測失敗時，並不意味着服務有問題，僅表示consul服務器和對應的rpc服務的端口連不上而已

rpc客戶端連接

客戶端需要向consul服務器請求，得到可用的rpc服務的ip，端口，在進行連接，我們還是基於之前的客戶端代碼進行修改，完整代碼如下：

 import grpc
 from dns import resolver
 from dns.exception import DNSException
 
 import test_pb2_grpc
 import test_pb2
 
 # 連接consul服務，作為dns服務器
 consul_resolver = resolver.Resolver()
 consul_resolver.port = 8600
 consul_resolver.nameservers=["127.0.0.1"]
 
 
 def get_ip_port(server_name):
    '''查詢出可用的一個ip，和端口'''
     try:
         dnsanswer = consul_resolver.query(f'{server_name}.service.consul', "A")
         dnsanswer_srv = consul_resolver.query(f"{server_name}.service.consul", "SRV")
     except DNSException:
         return None, None
     return dnsanswer[0].address, dnsanswer_srv[0].port
 
 ip, port = get_ip_port("order_server")
 
 channel = grpc.insecure_channel(f"{ip}:{port}")
 stub = test_pb2_grpc.OrderHandlerStub(channel)
 # 要完成請求需要先構造出proto文件中定義的請求格式
 ret = stub.create_order(test_pb2.OrderRequest(phone="12990", price="50"))
 
 print(ret.rst_string)

總結

至此，我們已經搭建好了一套簡單服務注冊-發現系統。當然consul的功能遠不止這一點，它還支持分布式，多節點部署。需要了解更多，可以去官網做進一步了解。

踩過的坑

• 在示例中，我們只捕獲了ctrl+c 信號，在實際中，服務會因為一些奇怪的問題宕掉，而此時，該節點信息會一直在consul里面，除非手動注銷，所以需要捕獲所有異常信號。
• 單節點的時候，consul服務掛掉會影響所有相關的項目，所以最好還是有從節點。