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

Question

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

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

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

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

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

Answer 1

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

我在grpc-python introduction中编辑了helloworld example来解释我的目的。

对于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;
}

我添加了Unsubsribe函数，允许一个特定的客户端与服务器建立连接。

在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，服务器可以区分不同的客户端，并向它们发送正确的相应信息。