这样的一段 grpc 客户端代码处理服务端流会不会有问题，或者有没有更优雅的实现方式

V2EX = way to explore

V2EX 是一个关于分享和探索的地方

已注册用户请登录

这是一个创建于 980 天前的主题，其中的信息可能已经有所发展或是发生改变。

public class gRPCWatcher {

    /**
     * grpc channel
     */
    private final gRPCChannel gRPCChannel;

    /**
     * 当前重连的次数
     */
    private int hasReconnectTimes = 0;

    /**
     * 最大重连次数
     */
    private final int maxReconnectTimes = 3;

    /**
     * 构造函数
     */
    public gRPCWatcher(gRPCChannel gRPCChannel) {
        this.gRPCChannel = gRPCChannel;
    }

    public void startWatch(Request confRequest, WatchHandler watchHandler) {
        StreamObserver<ConfResponse> streamObserver = new StreamObserver<ConfResponse>() {
            @Override
            public void onNext(ConfResponse confResponse) {
                try {
                    watchHandler.process(confResponse);
                } catch (Exception e) {
                    //...
                }
                hasReconnectTimes = 0;
            }

            @Override
            public void onError(Throwable throwable) {
                //error 的时候尝试重新调用，超过最大次数失败抛出异常
                if (hasReconnectTimes < maxReconnectTimes) {
                    hasReconnectTimes++;
                    startWatch(confRequest, watchHandler);
                } else {
                    throw new RuntimeException("reWatch "+maxReconnectTimes+" times still error" + throwable.getMessage());
                }
            }

            @Override
            public void onCompleted() {
                //...
            }
        };
        ConfigGrpc.newStub(gRPCChannel.getNettyChannel()).withWaitForReady().watch(confRequest, streamObserver);
    }
}

这是一段处理服务端流的 grpc 调用的逻辑，当网络出问题等异常情况的时候，会进入 onError, grpc channel 底层自带重连重试机制，所以我们只要重新 stub call 就行了。

所以我在 onError 的时候重新调用该方法，为了防止无限调，所以加了以这个最大重试次数，不知道这样写优雅不优雅- -

关于 grpc 的重连可以看官方的讨论，他们也是推荐 channel 可以复用主要 stub 重新 call https://groups.google.com/g/grpc-io/c/quToVM4NhdQ

3 条回复 • 2021-08-21 15:05:35 +08:00