go - 如何在 gRPC 中从服务器向客户端广播？

Question

我现在正在 gRPC 中创建一个小型聊天应用程序，我遇到了一个问题，如果用户想作为客户端连接到 gRPC 服务器，我想广播事件已经发生到所有其他连接的客户端。

我正在考虑使用某种观察者，但我对服务器如何知道谁已连接以及我如何将事件广播给所有客户端而不仅仅是一两个客户端感到困惑。

我知道使用流是部分答案，但因为每个客户端都在与服务器创建自己的流，所以我不确定它如何订阅其他服务器-客户端流。

Answer 1

另一种选择是使用长轮询方法。那就是尝试像下面这样的东西(Python 中的代码，因为这是我最熟悉的，但 go 应该非常相似)。这没有经过测试，只是为了让您了解如何在 gRPC 中进行长轮询:

.PROTO defs
-------------------------------------------------
service Updater {
    rpc GetUpdates(GetUpdatesRequest) returns (GetUpdatesResponse);
}

message GetUpdatesRequest {
    int64 last_received_update = 1;
}

message GetUpdatesResponse {
    repeated Update updates = 1;
    int64 update_index = 2;
}

message Update {
    // your update structure
}


SERVER
-----------------------------------------------------------
class UpdaterServer(UpdaterServicer):
    def __init__(self):
        self.condition = threading.Condition()
        self.updates = []

    def post_update(self, update):
        """
        Used whenever the clients should be updated about something. It will
        trigger their long-poll calls to return
        """
        with self.condition:
            # TODO: You should probably remove old updates after some time
            self.updates.append(updates)
            self.condition.notify_all()

    def GetUpdates(self, req, context):
        with self.condition:
            while self.updates[req.last_received_update + 1:] == []:
                self.condition.wait()
            new_updates = self.updates[req.last_received_update + 1:]
            response = GetUpdatesResponse()
            for update in new_updates:
                response.updates.add().CopyFrom(update)
            response.update_index = req.last_received_update + len(new_updates)
            return response


SEPARATE THREAD IN THE CLIENT
----------------------------------------------
request = GetUpdatesRequest()
request.last_received_update = -1
while True:
    stub = UpdaterStub(channel)
    try:
        response = stub.GetUpdates(request, timeout=60*10)
        handle_updates(response.updates)
        request.last_received_update = response.update_index
    except grpc.FutureTimeoutError:
        pass