grpc java：正确处理服务流调用的客户端重试

Question

我正在尝试在grpc上使用服务流和客户端上的异步存根建立一个简单的发布/订阅模式。在实现了部分流式消息返回到客户端之后，我想处理连接中断的场景。现在，我正在实现服务关闭时的部分，例如，客户端应该从连接丢失中“恢复”。

我在google/github/so上阅读并搜索了关于重试机制的内容，最后为流式传输消息的服务中的方法设置了重试策略。据我所知，当服务返回重试策略中定义的一些retryableStatusCodes时，重试机制应该可以工作。在客户端上引入重试策略后，我想对其进行测试，下面两个场景的结果让我对重试感到困惑。

第一种情况：

• 调用连接过程（有意在n秒后，没有消息流回到客户端）

第二种情况：

• 调用连接过程（第一条消息到达约n秒后，消息在客户端的onNext处理程序中处理）

总的来说，让我困惑的是，为什么这两种场景之间的行为有差异？为什么在第一个场景中检测到服务器返回不可用并尝试重试，但在第二个场景中，即使状态相同，重试也不起作用？

以下是客户端上的连接调用、服务上的连接方法以及客户端上重试策略的设置的代码

client:

messageStub.withWaitForReady().connect(messagesRequest, new StreamObserver<>() {
    @Override
    public void onNext(MessageResponse messageResponse) {
        //process new message
        MessageDto message = new MessageDto();
        message.setBody(messageResponse.getBody());
        message.setTitle(messageResponse.getTitle());

        messageService.broadcastMessage(message);
    }

    @Override
    public void onError(Throwable throwable) {
        //service went down
        LOGGER.error(throwable.getStackTrace());
    }

    @Override
    public void onCompleted() {
        //This method should be called when user logs out of the application
        LOGGER.info(String.format("Message streaming terminated for user %d", userId));
    }
});

service:

@Override
public void connect(MessageRequest request, StreamObserver<MessageResponse> responseObserver) {
    Long userId = request.getUserId();

    ServerCallStreamObserver<MessageResponse > serverCallStreamObserver =
        (ServerCallStreamObserver<MessageResponse >) responseObserver;
    serverCallStreamObserver.setOnCancelHandler(getOnCancelHandler(userId));
    registerClient(userId, serverCallStreamObserver);
    //responseObserver.onCompleted() is left out so connection is not terminated
}


@EventListener
public void listenForMessages(MessageEvent messageEvent) {
    //omitted code (just some data retrieving - populate conn and message vars)....

    MessageResponse.Builder builder = MessageResponse.newBuilder();
    StreamObserver<MessageResponse> observer = conn.getResponseObserver();
    builder.setType(message.getType());
    builder.setTitle(message.getTitle());
    builder.setBody(message.getBody());

    observer.onNext(builder.build())
}

retryPolicy:

{
  "methodConfig" : [
    {
      "name": [
        {
          "service": "ch.example.proto.MessageService",
          "method": "connect"
        }
      ],
      "retryPolicy": {
        "maxAttempts": 10,
        "initialBackoff": "5s",
        "maxBackoff": "30s",
        "backoffMultiplier": 2,
        "retryableStatusCodes": ["UNAVAILABLE"]
      }
    }
  ]
}

Answer 1

问题是，接收消息会提交RPC。这将在gRFC A6客户端重试中讨论。它提到了响应头，当服务器响应第一条消息时，会隐式发送这些头。

本质上，一旦gRPC将数据传递回客户端，就无法自动重试。如果gRPC重试，它应该如何将新流与它已经响应的流相结合？是否应跳过第一个响应？但如果现在的反应不同呢？元数据（通过响应头传递）的问题更为严重，因为这些元数据无法再次提供给客户端。

gRPC能够将客户端的请求重放到多个后端，但一旦它开始接收来自后端的响应，它将“固定”到该后端并且无法更改其决定。

您需要应用程序级重试才能重新建立流。当客户端重新建立流时，它可能需要修改请求以通知服务器客户端已经收到了哪些消息。