grpc server client callback 方式异步调用 Java 实现

在 Java 中实现 gRPC 的异步调用可以采用回调(callback)方式。下面是一个简单的示例：

1. 首先在服务端实现一个异步方法：

public class MyServiceImpl extends MyServiceGrpc.MyServiceImplBase {
    @Override
    public void myMethod(MyRequest request, StreamObserver<MyResponse> responseObserver) {
        // 异步执行任务
        ListenableFuture<MyResponse> future = executor.submit(() -> {
            // 执行任务
            MyResponse response = doTask(request);
            return response;
        });

        // 注册回调
        Futures.addCallback(future, new FutureCallback<MyResponse>() {
            @Override
            public void onSuccess(MyResponse response) {
                // 将结果返回给客户端
                responseObserver.onNext(response);
                responseObserver.onCompleted();
            }

            @Override
            public void onFailure(Throwable t) {
                // 异常处理
                responseObserver.onError(t);
            }
        });
    }

    private MyResponse doTask(MyRequest request) {
        // 执行任务
        return MyResponse.newBuilder().setMessage("Hello " + request.getName()).build();
    }
}

2. 在客户端异步调用服务端的方法：

public class MyClient {
    private final ManagedChannel channel;
    private final MyServiceGrpc.MyServiceStub stub;

    public MyClient(String host, int port) {
        channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build();
        stub = MyServiceGrpc.newStub(channel);
    }

    public void myMethodAsync(MyRequest request, StreamObserver<MyResponse> responseObserver) {
        // 发送请求
        stub.myMethod(request, responseObserver);
    }

    public void shutdown() throws InterruptedException {
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
    }
}

3. 最后，在客户端调用异步方法：

public class TestClient {
    public static void main(String[] args) throws Exception {
        MyClient client = new MyClient("localhost", 8888);

        // 异步调用
        CountDownLatch latch = new CountDownLatch(1);
        StreamObserver<MyResponse> responseObserver = new StreamObserver<MyResponse>() {
            @Override
            public void onNext(MyResponse response) {
                System.out.println("Received response: " + response.getMessage());
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("Error: " + t.getMessage());
                latch.countDown();
            }

            @Override
            public void onCompleted() {
                latch.countDown();
            }
        };
        client.myMethodAsync(MyRequest.newBuilder().setName("World").build(), responseObserver);

        // 等待异步调用完成
        latch.await();

        client.shutdown();
    }
}

以上就是一个简单的 gRPC 异步调用的实现示例，通过回调方式可以方便地处理异步调用的结果。