如果 grpc-python 服务器端有类实例，如何让每个客户端获取它们的状态？

Posted 2023-03-29

【中文标题】如果 grpc-python 服务器端有类实例，如何让每个客户端获取它们的状态？

【英文标题】：How to make each client get their state if there is class instance in grpc-python server side?

【发布时间】：2021-12-30 19:40:04

【问题描述】：

我想在下面的场景中使用grpc-python，但我不知道如何实现。

场景是，在python服务器中，它使用类来计算和更新实例的状态，然后将这个状态发送给相应的客户端；在客户端，需要多个客户端与服务器通信才能得到一个结果，而不受其他客户端的干扰。

具体来说，假设有一个类的初始值self.i=0，那么客户端每次调用该类的更新函数时，都会执行self.i=self.i+1，并返回self.i。实际上有两个客户端同时调用这个更新函数，比如client1第三次调用update，client2第一次调用update。

我认为这可以通过为每个客户端创建线程以避免冲突来解决。如果新客户端调用，则会创建新的客户端；如果现有客户端调用，将使用现有线程。但我不知道如何实现？

希望你能帮助我。提前致谢。

【问题讨论】：

到目前为止您尝试过什么？你的代码在哪里？

添加线程不会简化事情，反而会使事情复杂化。你需要独立线程做什么？如果两个客户端同时拨打电话，只需一个接一个地处理即可。

抱歉，我认为你没有理解我的问题。

【参考方案1】：

我想我自己解决了这个问题。如果您有其他更好的解决方案，可以在这里发布。

我在 grpc-python 简介中编辑了 helloworld 示例来解释我的目标。

对于 helloworld.proto

syntax = "proto3";

option java_multiple_files = true;
option java_package = "io.grpc.examples.helloworld";
option java_outer_classname = "HelloWorldProto";
option objc_class_prefix = "HLW";

package helloworld;

// The greeting service definition.
service Greeter 
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply) 
  rpc Unsubscribe (HelloRequest) returns (HelloReply) 


// The request message containing the user's name.
message HelloRequest 
  string name = 1;


// The response message containing the greetings
message HelloReply 
  string message = 1;

我添加了取消订阅功能以允许一个特定的客户端与服务器断开连接。

在 hello_server.py 中

import grpc
import helloworld_pb2
import helloworld_pb2_grpc
import threading
from threading import RLock
import time
from concurrent import futures
import logging


class Calcuate:
    def __init__(self):
        self.i = 0

    def add(self):
        self.i+=1
        return self.i


class PeerSet(object):
    def __init__(self):
        self._peers_lock = RLock()
        self._peers = 
        self.instances = 

    def connect(self, peer):
        with self._peers_lock:
            if peer not in self._peers:
                print("Peer  connecting".format(peer))
                self._peers[peer] = 1
                a = Calcuate()
                self.instances[peer] = a
                output = a.add()
                return output
            else:
                self._peers[peer] += 1
                a = self.instances[peer]
                output = a.add()
                return output


    def disconnect(self, peer):
        print("Peer  disconnecting".format(peer))
        with self._peers_lock:
            if peer not in self._peers:
                raise RuntimeError("Tried to disconnect peer '' but it was never connected.".format(peer))
            del self._peers[peer]
            del self.instances[peer]

    def peers(self):
        with self._peers_lock:
            return self._peers.keys()


class Greeter(helloworld_pb2_grpc.GreeterServicer):

    def __init__(self):
        self._peer_set = PeerSet()

    def _record_peer(self, context):
        return self._peer_set.connect(context.peer())

    def SayHello(self, request, context):
        output = self._record_peer(context)
        print("[thread ] Peers: , output: ".format(threading.currentThread().ident, self._peer_set.peers(), output))
        time.sleep(1)
        return helloworld_pb2.HelloReply(message='Hello, , !'.format(request.name, output))

    def Unsubscribe(self, request, context):
        self._peer_set.disconnect(context.peer())
        return helloworld_pb2.HelloReply(message=' disconnected!'.format(context.peer()))



def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    helloworld_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()


if __name__ == '__main__':
    logging.basicConfig()
    serve()

context.peer() 的使用改编自 Richard Belleville 的回答 in this post。您可以将 add() 函数更改为可用于更新实例状态的任何其他函数。

在 hello_client.py 中

from __future__ import print_function

import logging

import grpc
import helloworld_pb2
import helloworld_pb2_grpc


def run():
    # NOTE(gRPC Python Team): .close() is possible on a channel and should be
    # used in circumstances in which the with statement does not fit the needs
    # of the code.
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = helloworld_pb2_grpc.GreeterStub(channel)
        response = stub.SayHello(helloworld_pb2.HelloRequest(name='you'))
        print("Greeter client received: " + response.message)
        response = stub.SayHello(helloworld_pb2.HelloRequest(name='Tom'))
        print("Greeter client received: " + response.message)
        response = stub.SayHello(helloworld_pb2.HelloRequest(name='Jerry'))
        print("Greeter client received: " + response.message)
        stub.Unsubscribe(helloworld_pb2.HelloRequest(name="end"))


if __name__ == '__main__':
    logging.basicConfig()
    run()

如果我们同时运行多个 hello_client.py，服务器可以区分不同的客户端并发送正确的相应信息给它们。