grpc-python服务端存在类实例时如何让每个客户端获取独立状态？

你的核心诉求是不同客户端持有独立的计算状态、互不干扰，不需要单独给每个客户端绑定专属线程，更轻量的实现逻辑如下：

• 客户端请求时携带唯一身份标识（首次请求无标识的话，由服务端生成后返回给客户端后续复用）
• 服务端维护全局字典，key为客户端ID，value为对应客户端专属的计算类实例
• 处理请求时根据客户端ID读取对应实例执行计算逻辑，天然避免状态冲突，仅需要给全局字典的读写加线程锁避免并发插入冲突即可

1. 定义protobuf接口文件 #

新建counter.proto文件，内容如下：

syntax = "proto3";
package counter;

service CounterService {
  rpc Update (UpdateRequest) returns (UpdateResponse);
}

message UpdateRequest {
  optional string client_id = 1; // 首次请求可不传，后续复用服务端返回的ID
}

message UpdateResponse {
  string client_id = 1;
  int32 count = 2;
}

2. 生成gRPC相关代码 #

执行以下命令生成Python版的proto和gRPC代码：

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. counter.proto

3. 服务端实现 #

import grpc
import uuid
import threading
from concurrent import futures
import counter_pb2
import counter_pb2_grpc

# 你的计算类
class Counter:
    def __init__(self):
        self.i = 0
    
    def update(self):
        self.i += 1
        return self.i

class CounterServicer(counter_pb2_grpc.CounterServiceServicer):
    def __init__(self):
        self.client_instances = {} # 存储每个客户端的独立计算实例
        self.lock = threading.Lock() # 全局字典读写锁，避免并发插入冲突
    
    def Update(self, request, context):
        client_id = request.client_id
        # 首次请求无ID，生成新ID和对应实例
        if not client_id:
            with self.lock:
                client_id = str(uuid.uuid4())
                self.client_instances[client_id] = Counter()
        # 读取对应客户端的实例执行计算
        counter = self.client_instances[client_id]
        new_count = counter.update()
        return counter_pb2.UpdateResponse(client_id=client_id, count=new_count)

# 启动服务
def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    counter_pb2_grpc.add_CounterServiceServicer_to_server(CounterServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("服务端已启动，监听端口50051")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

4. 客户端实现 #

import grpc
import counter_pb2
import counter_pb2_grpc

class CounterClient:
    def __init__(self, host='localhost:50051'):
        self.channel = grpc.insecure_channel(host)
        self.stub = counter_pb2_grpc.CounterServiceStub(self.channel)
        self.client_id = None # 本地缓存服务端分配的ID
    
    def update(self):
        response = self.stub.Update(counter_pb2.UpdateRequest(client_id=self.client_id))
        self.client_id = response.client_id # 首次请求后保存ID，后续复用
        return response.count

# 测试逻辑
if __name__ == '__main__':
    # 模拟客户端1
    client1 = CounterClient()
    print(f"客户端1第1次调用: {client1.update()}") # 输出1
    print(f"客户端1第2次调用: {client1.update()}") # 输出2
    print(f"客户端1第3次调用: {client1.update()}") # 输出3

    # 模拟客户端2
    client2 = CounterClient()
    print(f"客户端2第1次调用: {client2.update()}") # 输出1
    print(f"客户端2第2次调用: {client2.update()}") # 输出2

• 你之前设想的为每个客户端创建独立线程的方案也可实现，但维护线程的开销远高于维护独立类实例，上述方案更轻量、性能更好
• 可给每个实例增加最后访问时间字段，定时清理长期无请求的实例，避免内存泄漏
• 多服务端实例部署场景下，可将实例状态存储到Redis等公共缓存，实现状态跨节点共享

内容的提问来源于stack exchange，提问作者Yanglan Wang