场景：使用ES过程中遇到Request cannot be executed; I/O reactor status: STOPPED 异常，意思是server端的连接异常终止了。项目使用了Spring data elasticsearch，会大量操作es。
原因：程序接口中将一块很大的数据存进JAVA集合中引发了oom，oom异常导致程序宕机，处于假死状态，进而导致 ES-CLIENT 和 ES-SERVER 端的 http 连接异常终止，SpringDataElasticsearch 和ES-SERVER 是长链接，只要报了OOM，当前和 ES-SERVER 的连接线程都将报异常，也就是说，虽然OOM只报了一次，但是可能有多个线程都在 Asserts.check 方法中报异常。
解决：低版本 httpClient 有这个问题，高版本就没有这个问题，是apache-httpclient 导致的。具体是因为捕获到线程异常之后http连接就主动终止了，解决思路是升级版本或者是重写处理器，遇到异常之后不终止连接。
升级httpclient相关pom依赖。
pom
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
<exclusions>
<exclusion>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-client</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-client</artifactId>
<version>7.15.2</version>
<exclusions>
<exclusion>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpcore</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpcore-nio</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<version>4.5.14</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpcore-nio</artifactId>
<version>4.4.16</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpcore</artifactId>
<version>4.4.16</version>
</dependency>
YAML
elasticsearch:
hosts:
- 192.168.0.1
port: 9200
username: anson
password: 123456
EsProperties
@Data
@ConfigurationProperties(prefix = "elasticsearch")
public class EsProperties {
private List<String> hosts;
private Integer port;
private String username;
private String password;
}
EsConnectionKeepAliveStrategy
public class EsConnectionKeepAliveStrategy extends DefaultConnectionKeepAliveStrategy {
public static final EsConnectionKeepAliveStrategy INSTANCE = new EsConnectionKeepAliveStrategy();
private EsConnectionKeepAliveStrategy() {
super();
}
/**
* 最大keep alive的时间（分钟）
* 这里默认为3分钟，可以根据实际情况设置。可以观察客户端机器状态为TIME_WAIT的TCP连接数，如果太多，可以增大此值。
*/
private final long MAX_KEEP_ALIVE_MINUTES = 3;
@Override
public long getKeepAliveDuration(HttpResponse response, HttpContext context) {
long keepAliveDuration = super.getKeepAliveDuration(response, context);
// <0 为无限期keepalive
// 将无限期替换成一个默认的时间
if(keepAliveDuration < 0){
return TimeUnit.MINUTES.toMillis(MAX_KEEP_ALIVE_MINUTES);
}
return keepAliveDuration;
}
}
EsClientConfig
@Slf4j
@Configuration
@EnableConfigurationProperties(EsProperties.class)
public class EsClientConfig {
@Autowired
private EsProperties properties;
@Bean
public ElasticsearchRestTemplate initElasticsearchRestTemplate() {
HttpHost[] hostArr = new HttpHost[properties.getHosts().size()];
for (int i = 0; i < properties.getHosts().size(); i++) {
hostArr[i] = new HttpHost(properties.getHosts().get(i), properties.getPort());
}
CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(properties.getUsername(), properties.getPassword()));
RestClientBuilder builder = RestClient.builder(hostArr)
.setRequestConfigCallback(
config -> config.setConnectTimeout(180000)
.setConnectionRequestTimeout(180000)
.setSocketTimeout(180000))
.setHttpClientConfigCallback(
httpClientBuilder -> {
httpClientBuilder.setMaxConnTotal(100);
httpClientBuilder.setMaxConnPerRoute(50);
httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
List<Header> headers = new ArrayList<>(2);
headers.add(new BasicHeader("Connection", "keep-alive"));
headers.add(new BasicHeader("Keep-Alive", "720"));
httpClientBuilder.setDefaultHeaders(headers);
httpClientBuilder.setKeepAliveStrategy(EsConnectionKeepAliveStrategy.INSTANCE);
try {
DefaultConnectingIOReactor ioReactor = new DefaultConnectingIOReactor();
ioReactor.setExceptionHandler(new IOReactorExceptionHandler() {
@Override
public boolean handle(IOException e) {
log.debug("System may be unstable: IOReactor encountered a checked exception : " + e.getMessage());
log.debug("start setHttpClientConfigCallback handle IOException e printStackTrace");
e.printStackTrace();
log.debug("end setHttpClientConfigCallback handle IOException e printStackTrace");
// Return true to note this exception as handled, it will not be re-thrown
return true;
}
@Override
public boolean handle(RuntimeException e) {
log.debug("System may be unstable: IOReactor encountered a runtime exception : " + e.getMessage() + ",e is {}", e);
log.debug("start setHttpClientConfigCallback handle RuntimeException e printStackTrace ");
e.printStackTrace();
log.debug("end setHttpClientConfigCallback handle RuntimeException e printStackTrace ");
// Return true to note this exception as handled, it will not be re-thrown
return true;
}
});
httpClientBuilder.setConnectionManager(new PoolingNHttpClientConnectionManager(ioReactor));
} catch (IOReactorException e) {
throw new RuntimeException(e);
}
return httpClientBuilder;
}
);
ElasticsearchRestTemplate elasticsearchRestTemplate = new ElasticsearchRestTemplate(new RestHighLevelClient(builder));
return elasticsearchRestTemplate;
}
}
ps. Keep Alive连接策略需换成自定义策略，否则会出现org.apache.catalina.connector.ClientAbortException: java.io.IOException: Broken pipe异常
网友评论