Camel http 组件不关闭连接 - Close_Wait答案

【问题标题】：Camel http component does not close connections - Close_WaitCamel http 组件不关闭连接 - Close_Wait
【发布时间】：2014-11-20 15:31:28
【问题描述】：

Camel http 组件没有正确关闭连接？

通过以下路线，我观察到正在服务器上创建连接，但没有终止。一段时间后，这会导致问题

java.io.IOException: Too many open files

路线：

from("seda:testSeda?concurrentConsumers=20")
    .setHeader("Connection", constant("Close"))
    .to("http://testServer/testFile.xml?authMethod=Basic&throwExceptionOnFailure=false&authUsername=user&authPassword=password")
    .to("file://abc")
.end();

连接处于Close_Wait状态有什么想法吗？

我在 2.14 版本中使用 camel-http 库

【问题讨论】：

您确定不是 Camel 客户端以外的其他进程导致了这个问题吗？在 Unix 上，您可以使用 netstat 来检查。
我确定。我将尝试切换到 http4 客户端，因为我可以看到它在我正在使用的 2.14 版本中支持 maxTotalConnections 和 connectionsPerRoute。我明天更新。
是的，camel-http 中可能缺少一些东西。我认为我们没有检测到最终用户是否已将 Connection 标头设置为 Close。随时登录 JIRA 票证：camel.apache.org/support
我刚刚查看了HttpProducer的代码，它在处理方法结束时调用了method.releaseConnection()。
即使使用 ProducerTemplate 和 HTTP 组件，连接也不会被释放。请提出建议。

标签： java apache-camel

【解决方案1】：

您可以覆盖 Apache Camel 使用的默认 HttpClient 并定义自定义的 Keep Alive 策略。

https://howtodoinjava.com/spring-boot2/resttemplate/resttemplate-httpclient-java-config/

下面的代码解决了我在生产中的问题：

@配置公共类 AppConfiguration {

@Autowired
private PoolingHttpClientConnectionManager poolingConnectionManager;
@Autowired
private ConnectionKeepAliveStrategy connectionKeepAliveStrategy;
@Autowired
private SSLConnectionSocketFactory sslContext;

@Bean
CamelContextConfiguration contextConfiguration() {
    return new CamelContextConfiguration() {
        @Override
        public void beforeApplicationStart(CamelContext context) {
            HttpComponent httpComponent = context.getComponent("https4", HttpComponent.class);

            httpComponent.setHttpClientConfigurer(new HttpClientConfigurer() {
                @Override
                public void configureHttpClient(HttpClientBuilder builder) {

                    builder.setSSLSocketFactory(sslContext);

                    RegistryBuilder.<ConnectionSocketFactory>create().register("https", sslContext).build();

                    builder.setConnectionManager(poolingConnectionManager);
                    builder.setKeepAliveStrategy(connectionKeepAliveStrategy);
                }
            });
        }

        @Override
        public void afterApplicationStart(CamelContext arg0) {

        }

    };
}

}

@配置公共类 HttpClientConfig {

private static final int DEFAULT_KEEP_ALIVE_TIME_MILLIS = 20 * 1000;
private static final int CLOSE_IDLE_CONNECTION_WAIT_TIME_SECS = 30;

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

private Logger logger = LoggerFactory.getLogger(HttpClientConfig.class);

@Bean
public PoolingHttpClientConnectionManager poolingConnectionManager() {
    PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
    connectionManager.setMaxTotal(200);
    connectionManager.setDefaultMaxPerRoute(20);

    return connectionManager;
}

@Bean
public CloseableHttpClient httpClient() {
    RequestConfig config = RequestConfig.custom().setConnectionRequestTimeout(5000).setConnectTimeout(5000)
            .setSocketTimeout(15000).build();

    return HttpClientBuilder.create().setSSLSocketFactory(this.getSSLContext())
            .setConnectionManager(this.poolingConnectionManager()).setDefaultRequestConfig(config)
            .setKeepAliveStrategy(this.connectionKeepAliveStrategy()).build();

}

@Bean
public ConnectionKeepAliveStrategy connectionKeepAliveStrategy() {
    return new ConnectionKeepAliveStrategy() {
        @Override
        public long getKeepAliveDuration(HttpResponse response, HttpContext context) {
            HeaderElementIterator it = new BasicHeaderElementIterator(
                    response.headerIterator(HTTP.CONN_KEEP_ALIVE));
            while (it.hasNext()) {
                HeaderElement he = it.nextElement();
                String param = he.getName();
                String value = he.getValue();

                if (value != null && param.equalsIgnoreCase("timeout")) {
                    return Long.parseLong(value) * 1000;
                }
            }
            return DEFAULT_KEEP_ALIVE_TIME_MILLIS;
        }
    };
}

@Bean
public Runnable idleConnectionMonitor(final PoolingHttpClientConnectionManager connectionManager) {
    return new Runnable() {
        @Override
        @Scheduled(fixedDelay = 10000)
        public void run() {
            if (connectionManager != null) {
                connectionManager.closeExpiredConnections();
                connectionManager.closeIdleConnections(CLOSE_IDLE_CONNECTION_WAIT_TIME_SECS, TimeUnit.SECONDS);
            }
        }
    };
}

@Bean
public SSLConnectionSocketFactory getSSLContext() {
    try {
        KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());

        try (FileInputStream jksFile = new FileInputStream(this.pathCertificado)) {
            keyStore.load(jksFile, "xxxxxx".toCharArray());
        }

        TrustStrategy acceptingTrustStrategy = (X509Certificate[] chain, String authType) -> true;

        SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(keyStore, acceptingTrustStrategy).build();

        return new SSLConnectionSocketFactory(sslContext);
    } catch (Exception e) {
        logger.error("Keystore load failed: " + this.pathCertificado, e);

        return null;
    }
}

}

【讨论】：