配置Spring WebFlux WebClient以使用自定义线程池

Question

是否可以将WebClient配置为使用Reactor-http-nio线程池以外的自定义线程池（使用Netty时？如果可能，我们是否可以以某种方式限制该自定义线程池仅在特定处理器内核上运行？

Answer 1

是。您可以。

1. 在您自己的线程池和EventLoopGroup中创建一些（或创建NioEventLoopGroup bean）。例如：

   {
    Intger THREADS = 10;
   
    BasicThreadFactory THREADFACTORY = new BasicThreadFactory.Builder()
           .namingPattern("HttpThread-%d")
           .daemon(true)
           .priority(Thread.MAX_PRIORITY)
           .build();
   
    EXECUTOR = new ThreadPoolExecutor(
           THREADS,
           THREADS,
           0L,
           TimeUnit.MILLISECONDS,
           new LinkedBlockingQueue<>(),
           THREADFACTORY,
           new ThreadPoolExecutor.AbortPolicy());
   
    NioEventLoopGroup RESOURCE= new NioEventLoopGroup(THREADS,EXECUTOR);
   }
   

2. 注册您自己的ReactorResourceFactory。并提供基于自定义线程执行器的自己的EventLoopGrooup]

   @Bean
   public ReactorResourceFactory reactorResourceFactory(NioEventLoopGroup RESOURCE) {
       ReactorResourceFactory f= new ReactorResourceFactory();
       f.setLoopResources(new LoopResources() {
           @Override
            public EventLoopGroup onServer(boolean b) {
                return RESOURCE;
               }
           });
       f.setUseGlobalResources(false);
       return f;
   }
   

3. 然后注册ReactorClientHttpConnector。在下面的示例中，使用了自定义SSL上下文

   @Bean
   public ReactorClientHttpConnector reactorClientHttpConnector(ReactorResourceFactory r) throws SSLException {
       SslContext sslContext = SslContextBuilder
           .forClient()
           .trustManager(InsecureTrustManagerFactory.INSTANCE)
           .build();
       return new ReactorClientHttpConnector(r, m -> m.secure(t -> t.sslContext(sslContext)));
   }
   

4. 最终构建WebClient

   @Bean
   public WebClient webClient(ReactorClientHttpConnector r) {
       return WebClient.builder().clientConnector(r).build();
   }
   

如果要对WebServer使用相同的内容。对ReactiveWebServerFactory进行相同的配置。

    @Bean
    public ReactiveWebServerFactory reactiveWebServerFactory(NioEventLoopGroup RESOURCE) {
        NettyReactiveWebServerFactory factory = new NettyReactiveWebServerFactory();
        factory.addServerCustomizers(hs->hs.tcpConfiguration(s->s.runOn(RESOURCE)));
        return factory;
    }

进口：

    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.handler.ssl.SslContext;
    import io.netty.handler.ssl.SslContextBuilder;
    import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.DependsOn;
    import org.springframework.http.client.reactive.ReactorClientHttpConnector;
    import org.springframework.http.client.reactive.ReactorResourceFactory;
    import org.springframework.stereotype.Component;
    import org.springframework.web.reactive.function.client.WebClient;
    import reactor.netty.resources.LoopResources;
    import org.apache.commons.lang3.concurrent.BasicThreadFactory;
    import org.springframework.boot.web.embedded.netty.NettyReactiveWebServerFactory;
    import org.springframework.boot.web.reactive.server.ReactiveWebServerFactory;
    import java.util.concurrent.*;