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

Question

我想在下面的场景中使用grpc-python，但是不知道怎么实现

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

具体来说，假设有一个class，初始值self.i =0，那么每次客户端调用class的更新函数时，它都会self.i=self.i+1 和 returns self.i。实际上有两个client同时调用了update函数，比如client1第三次调用update，client2第一次调用update

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

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

Answer 1

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

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

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

How to make each client get their state if there is class instance in grpc-python server side?

python

multithreading

grpc

grpc-python