WebFlux 中的 fromCallable 封装了多个阻塞调用

Question

我在反应式 Java 中使用 Feign Client。 Feign 客户端有一个拦截器，它发送阻塞请求以获取身份验证令牌，并将其作为标头添加到 feign 请求中。

feign 请求通过 Schedulers.boundedElastic() 封装在 Mono.FromCallable 中。

我的问题是：获取身份验证令牌的内部调用是否被视为阻塞调用？

我知道这两个调用将位于与 Schedulers.boundedElastic() 不同的线程上，但不确定是否可以在同一线程上执行它们，或者我应该更改它，以便它们在不同的线程上运行。

Feign客户端：

@FeignClient(name = "remoteRestClient", url = "${remote.url}",
        configuration = AuthConfiguration.class, decode404 = true)
@Profile({ "!test" })
public interface RemoteRestClient {
    @GetMapping(value = "/getSomeData" )
    Data getData();
}

拦截器：

public class ClientRequestInterceptor implements RequestInterceptor {

    private IAPRequestBuilder iapRequestBuilder;
    private String clientName;

    public ClientRequestInterceptor(String clientName, String serviceAccount, String jwtClientId) {
        this.iapRequestBuilder = new IAPRequestBuilder(serviceAccount, jwtClientId);
        this.clientName = clientName;
    }

    @Override
    public void apply(RequestTemplate template) {
        try {
            HttpRequest httpRequest = iapRequestBuilder.buildIapRequest();   <---- blocking call
            template.header(HttpHeaders.AUTHORIZATION, httpRequest.getHeaders().getAuthorization());
        } catch (IOException e) {
            log.error("Building an IAP request has failed: {}", e.getMessage(), e);
            throw new InterceptorException(String.format("failed to build IAP request for %s", clientName), e);
        }
    }
}

feign配置：

public class AuthConfiguration {
    @Value("${serviceAccount}")
    private String serviceAccount;

    @Value("${jwtClientId}")
    private String jwtClientId;

    @Bean
    public ClientRequestInterceptor getClientRequestInterceptor() {
        return new ClientRequestInterceptor("Entitlement", serviceAccount, jwtClientId);
    }
}

和feign客户端调用：

  private Mono<Data> getData() {
return Mono.fromCallable(() -> RemoteRestClient.getData()
                                .publishOn(Schedulers.boundedElastic());
    }

Answer 1

您可以看出这是一个阻塞调用，因为它返回一个具体类而不是 Future （Mono 或 Flux）。为了能够返回具体的类，线程需要等待，直到我们得到返回它的响应。

所以是的，这很可能是一个阻塞调用。

Reactor 建议您使用 subscribeOn< /code> 运算符在执行阻塞调用时，这会将整个运算符链放置在其自己的线程池上。

您已选择使用 publishOn ，并且值得指出 文档：

影响后续运算符的执行位置

这实际上意味着在 publishOn 运算符之前，所有操作都将使用任何可用的匿名线程执行。

但是之后的所有调用都将在定义的线程池上执行。

private Mono<Data> getData() {
    return Mono.fromCallable(() -> RemoteRestClient.getData()
                                .publishOn(Schedulers.boundedElastic());
}

您已选择将其放置在之后，以便线程池切换将在调用getData之后完成。

publishOn 在链中的放置很重要，而 subscribeOn 会影响整个操作符链，这意味着它的放置并不重要。

因此，再次回答您的问题，是的，这很可能是一个阻塞调用（我无法 100% 确认，因为我没有查看源代码）以及您希望如何使用 publishOn subscribeOn 取决于您。

或者看看是否有一个反应式替代库可供使用。