• • 已关闭**。此问题为opinion-based。当前不接受答案。
• • 想要改进此问题吗？**请更新此问题，以便editing this post可以用事实和引文来回答。

6小时前关门了。
Improve this question
我有一个通过grpc的双向频道

@Override
public StreamObserver<MessageReceiver.MessageRequest> biDirectionalMessageStream(StreamObserver<MessageReceiver.MessageResponse> responseObserver) {
    return new StreamObserver<>() {
        @Override
        public void onNext(MessageReceiver.MessageRequest messageRequest) {
            responseObserver.onNext(MessageReceiver.MessageResponse.newBuilder()
                    .setMessage("SomeMessage")
                    .build());
        }

        @Override
        public void onError(Throwable throwable) {
            responseObserver.onError(throwable);
        }

        @Override
        public void onCompleted() {
            responseObserver.onCompleted();
        }
    };
}

消息源是Kafka。本主题中的消息是针对不同的grpc连接的。在收到grpc消息时，连接可能尚未打开。如何在打开grpc连接时传输打开连接之前收到的所有消息，然后从Kafka实时传输这些消息？
我看到几个选项：
1.在内存中创建一个缓冲区，将来自kafka的消息读取到该缓冲区中，并在打开连接时将消息从缓冲区发送到grpc通道
1.建立持久存储（redis/mongo/postgres等）并使用它代替缓冲区
不确定哪个解决方案更好（如果是后者，不确定选择哪个存储）。负载配置文件-17，000 rpc