JobsV1Beta3Client 响应式编程常见问题解答：如何解决 `fetchActiveJobs()` 方法抛出 `UNAVAILABLE` 错误？

2024-04-15 03:07:46

GCP SDK 中 JobsV1Beta3Client 响应式编程常见问题疑难解答

简介

对于使用响应式编程模式的开发者来说，在使用 GCP Java SDK's JobsV1Beta3Client 类获取所有正在运行的数据流作业时，经常会遇到一个问题：fetchActiveJobs() 方法会抛出 com.google.api.gax.rpc.UnavailableException: io.grpc.StatusRuntimeException: UNAVAILABLE: Channel shutdown invoked 错误。本文将深入探讨导致此问题的原因，并提供一个全面的解决方案。

问题根源

为了理解这个问题，我们需要深入了解 gRPC 的工作原理。gRPC 是一个用于在客户端和服务器之间进行远程过程调用的框架。gRPC 客户端在内部维护一个连接池，用于向服务器发送请求。当使用响应式风格时（例如 Mono.fromCallable()），客户端不会立即执行请求，而是将其订阅到响应式流中。这种做法导致客户端连接池中的连接保持打开状态的时间比使用阻塞风格时长得多，从而导致连接因超时而关闭。

解决方案

要解决此问题，需要在执行请求之前显式地关闭客户端。这可以通过使用 try-with-resources 块来实现：

public Mono<JobsV1Beta3Client.ListJobsPage> fetchActiveJobs() {
    return Mono.fromCallable(() -> {
        try (final JobsV1Beta3Client jobsV1Beta3Client = JobsV1Beta3Client.create()) {
            var request = ListJobsRequest.newBuilder()
                    .setProjectId("my-project-id")
                    .setLocation("my-region")
                    .setFilter(ListJobsRequest.Filter.ACTIVE)
                    .build();

            return jobsV1Beta3Client.listJobs(request);
        }
    })
    .map(AbstractPagedListResponse::getPage)
    .subscribeOn(Schedulers.boundedElastic());
}