go使用grpc实现异步_(python、go)基于ETCD的gRPC分布式服务器实现详解

作者：Zarten知乎专栏：框架工具篇详解知乎ID： Zarten简介：互联网一线工作者，尊重原创并欢迎评论留言指出不足之处，也希望多些关注和点赞是给作者最好的鼓励！

1-概述

gRPC框架是一个性能很好的rpc框架，但框架中没有实现分布式服务器负载均衡的方法，只是给出实现方案，需要我们自己实现。官方推荐是客户端负载均衡的方案，也就是由客户端主动选择路由，这样的好处是不用给代理服务器造成压力。

分布式场景下一般采用etcd、consul、zookeeper等分布式系统，这里采用etcd，etcd是go语言实现，在github可看到开源的全部实现代码。

目前网上都是go语言写的关于基于etcd的grpc分布式服务器的实现，几乎没有python版的grpc服务器实现，于是本人就写了一个python版的服务器和不同客户端（python、go）版的全套实现方案。

注：全部完整代码已上传至github：Zartenc/grpc_etcd_ms_py

2-实现思想

2-1-服务器端实现思想

服务器端思想主旨是：每个gRPC服务器启动（上线）都会在etcd的key中注册自己在本机中对外暴露的ip-port，一旦gRPC服务停止（下线、意外挂掉、租约到期等）会在etcd中注销掉自己的信息。

2-2-客户端实现思想

客户端思想主旨是：只需连接到etcd服务器产生一个etcdClient对象，长期维护这个对象即可，通过这个对象的负载均衡策略可获取众多gRPC服务器中的其中一个进行连接。在这个etcdClient对象内部会对etcd的key添加一个监视（watch）并维护一个可用gRPC服务器信息的集合。

3-具体实现核心代码

前提已部署并启动etcd服务器，若没部署请参考这里。

3-1-服务器端实现核心代码

服务器端采用python编写。

由于采用etcd分布式框架，这里首先实现一个etcd客户端的类EtcdClient，类方法有:

get_values_by_key():通过etcd的key获取服务器信息。
put_values_by_key():添加服务器信息到etcd的key中。

class EtcdClient(etcd3.Etcd3Client):def get_values_by_key(self, key, **kwargs):values, _ = self.get(key, **kwargs)values_list = []if values is not None:try:values_list = json.loads(values.decode('utf-8'))if not isinstance(values_list, list):raise TypeError()except:raise Exception()return values_listdef put_values_by_key(self, key, values):if not isinstance(values, list):raise Exception()self.put(key, json.dumps(values))

再实现一个此服务器对etcd操作的类EtcdHandleServ，类方法有：

register_service():注册本机信息到etcd的key中。

logout_service:从etcd的key中注销本机信息。

class EtcdHandleServ():def __init__(self, service_port, etcd_ip, etcd_port, etcd_prefix):self.etcd_ip = etcd_ipself.etcd_port = etcd_portself.etcd_prefix = etcd_prefix# service_ip = get_outside_ip()service_ip = '127.0.0.1' #在本机机器作实验使用self.endpoint = f'{service_ip}:{service_port}'def register_service(self):etcd_client = EtcdClient(host=self.etcd_ip, port=self.etcd_port)key_name = f'{self.etcd_prefix}/grpc'with etcd_client.lock(key_name):value_list = etcd_client.get_values_by_key(key_name)if self.endpoint not in value_list:value_list.append(self.endpoint)etcd_client.put_values_by_key(key_name, value_list)def logout_service(self):etcd_client = EtcdClient(host=self.etcd_ip, port=self.etcd_port)key_name = f'{self.etcd_prefix}/grpc'with etcd_client.lock(key_name):value_list = etcd_client.get_values_by_key(key_name)if self.endpoint in value_list:value_list.remove(self.endpoint)etcd_client.put_values_by_key(key_name, value_list)

最后在主函数中进行相关的注册和注销操作并监控程序停止信号。

*注意：在docker启动的操作事项

在下面代码中，若服务器要在docker中启动需要考虑2个问题：

1.在docker run命令启动时如何将参数从外部传到容器内？
2.docker stop命令停止时如何程序内部如何接收停止信号？

上面2个问题共同之处是在Dockerfile文件中的ENTRYPOINT命令：

ENTRYPOINT ["python3", "-u", "main.py"]

在Dockerfile文件中启动命名使用ENTRYPOINT的exec模式，这样程序在容器内为1号进程，可接收停止信号（若为shell模式，也可处理，但麻烦一些，后面给出解决方案）。

docker启动时从外部传参，只需在后面跟上需要传入的参数即可：

docker run -d -p 65510:65510 zartenImage:v ----service_port 65510 --etcd_ip xxxx --etcd_prefix /zarten

上面给出了2个问题的解决方案，具体更全面的方案，请参考本人之前写的一篇docker使用详解的文章，并在文章最后的常见问题中有提到解决方案。

def main(service_ip, service_port, etcd_ip, etcd_port, etcd_prefix):print('***service is starting...')grpc_server = grpc.server(futures.ThreadPoolExecutor(max_workers=500))zarten_pb2_grpc.add_ZartenServicer_to_server(ZartenServ(), grpc_server)grpc_server.add_insecure_port(f'{service_ip}:{service_port}')grpc_server.start()etcd_handle_serv = EtcdHandleServ(service_port=service_port, etcd_ip=etcd_ip, etcd_port=etcd_port, etcd_prefix=etcd_prefix)etcd_handle_serv.register_service()event = threading.Event()def signal_handler(*args):etcd_handle_serv.logout_service()event.set()signal.signal(signal.SIGINT, signal_handler)signal.signal(signal.SIGTERM, signal_handler)print("***serveice started")try:while True:time.sleep(60 * 60 * 24)except KeyboardInterrupt:etcd_handle_serv.logout_service()grpc_server.stop(0)

最后采用命令方式启动。

# python main.py --service_port 65510 --etcd_ip xxxx --etcd_prefix /zarten
if __name__ == '__main__':parser = get_arguments_parser()args = parser.parse_args()main(args.service_ip, args.service_port, args.etcd_ip, args.etcd_port, args.etcd_prefix)

3-2-客户端实现核心代码

客户端采用python和go编写。也可采用其他语言实现，这里就不作展示了，知道实现思想即可自己实现。

gRPC客户端中的负载均衡是以每次调用为基础而不是以每个连接为基础，即只需维护一个连接对象，每次调用都是连接不同gRPC的服务器。官方文档是这样描述的：

“It is worth noting that load-balancing within gRPC happens on a per-call basis, not a per-connection basis. In other words, even if all requests come from a single client, we still want them to be load-balanced across all servers.”

3-2-1-python客户端

首先同样是一个连接etcd的客户端类，跟服务器端代码差不多。

class EtcdClient(etcd3.Etcd3Client):def get_values_by_key(self, key, **kwargs):values, _ = self.get(key, **kwargs)values_list = []if values is not None:try:values_list = json.loads(values.decode('utf-8'))if not isinstance(values_list, list):raise TypeError()except:raise Exception()return values_listdef put_values_by_key(self, key, values):if not isinstance(values, list):raise Exception()self.put(key, json.dumps(values))

其次是此客户端对etcd的操作类，采用单例模式，主要方法就是监视etcd并进行回调处理。

class EtcdHandleClient(EtcdClient):_singleton = Nonedef __new__(cls, *args, **kwargs):if not cls._singleton:cls._singleton = super().__new__(cls)return cls._singletondef __init__(self, etcd_ip, etcd_port, etcd_prefix):self.etcd_ip = etcd_ipself.etcd_port = etcd_portself.etcd_prefix = etcd_prefixsuper().__init__(host=etcd_ip, port=etcd_port)self.endpoints_list = self.get_values_by_key(f'{self.etcd_prefix}/grpc')self.watched_id = self.add_watch_callback(key=f'{self.etcd_prefix}/grpc', callback=self._update_endpoints)def __del__(self):self.cancel_watch(self.watched_id)def get_grpc_serv_ip(self):endpoints_nums = len(self.endpoints_list)if endpoints_nums <= 0:raise RuntimeError('No grpc services are available.Please notify the administrator to start the grpc service')select_id = random.randint(0, len(self.endpoints_list)-1)return self.endpoints_list[select_id]def _update_endpoints(self, watched_response):watched_event = watched_response.events[0]try:update_endpoint_list = json.loads(watched_event.value)if not isinstance(update_endpoint_list, list):raise TypeErrorexcept Exception as e:print(e)returnself.endpoints_list = update_endpoint_list

最后main函数中只需长期维护一个EtcdHandleClient对象即可。

def main():etcd_client = EtcdHandleClient(etcd_ip='xxxx', etcd_port=2379, etcd_prefix='/zarten')endpoint = etcd_client.get_grpc_serv_ip()print('endpoint:', endpoint)with grpc.insecure_channel(endpoint) as channel:stub = zarten_pb2_grpc.ZartenStub(channel)response = stub.GetInfo(zarten_pb2.ZartenRequest(zhihu_name='Zarten123'))print(f'receive response: {response}')

3-2-2-go客户端

go语言客户思想跟python一样，只是代码不同而已。

首先定义一个GrpcClient结构体，包括一个etcd连接对象和一个可用gRPC服务器信息数组。

type GrpcClient struct {Etcd3Client *clientv3.ClientGrpcEndpoints []string
}

其次是初始化GrpcClient结构体的函数NewGrpcClient()，此函数中会调用一个协程来监视etcd的变动。

GrpcClient结构体的方法只有一个GetRrpcServIp()对外开放函数来获取某个gRPC服务器的信息。

func NewGrpcClient(EtcdIp string, EtcdPort int, EtcdPrefix string) *GrpcClient{keyName := EtcdPrefix+"/grpc"grpcClient := new(GrpcClient)cli, err := clientv3.New(clientv3.Config{Endpoints:   []string{EtcdIp + ":" + strconv.Itoa(EtcdPort)},DialTimeout: 10 * time.Second,})if err != nil {log.Fatal(err)}res, err := cli.Get(context.Background(), keyName)if err != nil{log.Fatal(err)}for _, ev := range res.Kvs {endPoints := ev.Valueerr := json.Unmarshal(endPoints, &grpcClient.GrpcEndpoints)if err != nil{log.Fatal(err)}break}if len(grpcClient.GrpcEndpoints) <= 0{log.Fatal("No grpc services are available.Please notify the administrator to start the grpc service")}grpcClient.Etcd3Client = clirch := cli.Watch(context.Background(), keyName)go func() {for wresp := range rch {for _, ev := range wresp.Events {fmt.Printf("%s %q : %qn", ev.Type, ev.Kv.Key, ev.Kv.Value)mu.Lock()err := json.Unmarshal(ev.Kv.Value, &grpcClient.GrpcEndpoints)mu.Unlock()if err != nil{log.Fatal(err)}fmt.Println(grpcClient.GrpcEndpoints)}}}()return grpcClient
}func (g *GrpcClient) GetRrpcServIp() string{rand.Seed(time.Now().Unix())n := len(g.GrpcEndpoints)return g.GrpcEndpoints[rand.Intn(n)]
}

main函数中只需维护一个NewGrpcClient对象即可。

func main() {grpcClient := client_center.NewGrpcClient("xxxx", 2379, "/zarten")ip := grpcClient.GetRrpcServIp()fmt.Println(ip)conn, err := grpc.Dial(grpcClient.GetRrpcServIp(), grpc.WithInsecure(), grpc.WithBlock())if err != nil {log.Fatalf("did not connect: %v", err)}defer conn.Close()client := go_protoc.NewZartenClient(conn)res, _ := client.GetInfo(context.Background(), &go_protoc.ZartenRequest{ZhihuName:"zarten456"})fmt.Println(res.Name)fmt.Println(res.Homepage)}

4-演示结果

在本地机器启动3个服务器端，端口分别为65510、65511、65512.

多次调用python和go版的客户端，都是使用不同gRPC服务器并成功返回信息。部分截图如下所示：

5-总结

主旨思想是通过etcd来进行交互来共享所有的gRPC服务器信息。

在服务器端的实现为了能更加健全，还需考虑加入租约功能，此功能主要用于服务器端自身挂掉后无法及时通知etcd来注销掉自己信息。此功能以后有时间会在github中更新，敬请期待！