AsyncRestTemplate 配置队列大小

Posted 2023-02-19

【中文标题】AsyncRestTemplate 配置队列大小

【英文标题】：AsyncRestTemplate config queue size

【发布时间】：2017-06-07 01:14:19

【问题描述】：

我有一个每小时运行的作业，在 40core 服务器上，每个作业可以有 1 到 10 万个任务（需要大队列），每个任务在完成时执行 HTTP 请求，每个任务都很关键，这意味着它必须运行并完成。

任务可以异步运行。

如何配置池中的线程数？如何配置 queue 大小？

在这个测试中，我试图让我的任务被拒绝并淹没我的线程池，但我却得到了 SocketTimeoutException

public static void main(String[] args) throws IOReactorException 
    String url = "http://internal.server:8001/get";
    int connectionTimeout = 3000;
    int soTimeout = 3000;
    int maxHttpConnections = 30;

    IOReactorConfig customIOReactorConfig = IOReactorConfig.custom()
            .setIoThreadCount(Runtime.getRuntime().availableProcessors())
            .setConnectTimeout(connectionTimeout)
            .setSoTimeout(soTimeout)
            .build();

    ConnectingIOReactor ioReactor = new DefaultConnectingIOReactor(customIOReactorConfig);

    PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
    connManager.setDefaultMaxPerRoute(maxHttpConnections);
    connManager.setMaxTotal(maxHttpConnections);
    CloseableHttpAsyncClient customHttpAsyncClient = HttpAsyncClients.custom()
            .setConnectionManager(connManager)
            .build();

    HttpComponentsAsyncClientHttpRequestFactory asyncRequestFactory = new HttpComponentsAsyncClientHttpRequestFactory(customHttpAsyncClient);
    AsyncRestTemplate asyncRestTemplate = new AsyncRestTemplate(asyncRequestFactory);

    System.out.println("start");
    for (int i = 0; i < 30_000; i++) 
        asyncRestTemplate.execute(url, HttpMethod.GET, request -> logger.info("doWithRequest..."), response -> 
            logger.info("extractData...");
            return response.getStatusText();
        ).addCallback(new ListenableFutureCallback<String>() 
            @Override
            public void onFailure(Throwable ex) 
                logger.error("onFailure [] []", ex.getMessage(), ex.getStackTrace()[0].toString());
            

            @Override
            public void onSuccess(String result) 
                logger.info("onSuccess");
            
        );
    
    System.out.println("end loop");

【参考方案1】：

你可以这样做：

ThreadPoolTaskExecutor poolTaskExecutor = new ThreadPoolTaskExecutor();
poolTaskExecutor.setQueueCapacity(100);
CloseableHttpAsyncClient httpclient = HttpAsyncClients
     .custom()
     .setThreadFactory(poolTaskExecutor).build();