搜索网站logo怎么做,wordpress定时发布没用,河南省建设教育协会网站,高仿卡西欧手表网站Java向ES库中插入数据报错#xff1a;java.lang.IllegalStateException: Request cannot be executed; I/O reactor status: STO 一、问题问题原因 二、解决思路 一、问题
在使用Java向ES库中插入数据时#xff0c;第一次成功插入#xff0c;第二次出现以下错误#xff1a… Java向ES库中插入数据报错java.lang.IllegalStateException: Request cannot be executed; I/O reactor status: STO 一、问题问题原因 二、解决思路 一、问题
在使用Java向ES库中插入数据时第一次成功插入第二次出现以下错误 java.lang.IllegalStateException: Request cannot be executed; I/O reactor status: STOPPED at
问题原因
这里显示是连接中断第一次遇到这个问题比较疑惑为什么es的客户端会中断理论上es client 是长连接不停的有数据写入连接一致存在除非有服务端异常。在elasticsearch服务端查看日志没有任何异常信息。
网上搜索错误信息原来是 Apache HTTPComponents 异步客户端问题。es官网有个issues详细的记录的这个问题的原因和修复建议。
官方的意见是 Apache HTTPComponents 异步客户端 使用了一个内部的I/O reactor 分发IO event。在某些情况下IO reactor会记录程序调用栈中的异常或者Java NOI库中的异常如果这些异常不被处理I/O reactor会直接关闭es client不可用此时只能重启服务。es client中试图增加一个默认的 I/O reactor 异常处理逻辑但是在做了一些尝试后发现捕获I/O reactor后会导致SSL中断。而HTTPComponents 在版本5中已经修复了这个问题最终官网给的建议是等待版本升级。
二、解决思路
解决问题的过程中参考了以下文档 https://www.cnblogs.com/yangchongxing/p/15440197.html https://github.com/elastic/elasticsearch/issues/42133 https://zhuanlan.zhihu.com/p/384269417 https://cloud.tencent.com/developer/article/1806886
主要获得解决方法的是以下 https://github.com/elastic/elasticsearch/issues/39946 主要引用以下 大概意思是说: 在每个线程需要时创建一个新的客户端并在方法结束时关闭。这就解决了问题。 结合GPT获取解决方案
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;public class EsClientManager {private final ExecutorService executorService Executors.newFixedThreadPool(10);private final ThreadLocalRestHighLevelClient CLIENT_THREAD_LOCAL ThreadLocal.withInitial(() - {RestHighLevelClient client null;try {client RestHighLevelClient.builder(new HttpHost(localhost, 9200, http)).build();return client;} catch (IOException e) {e.printStackTrace();return null;}});public RestHighLevelClient getClient() {return CLIENT_THREAD_LOCAL.get();}public void closeClient(RestHighLevelClient client) {CLIENT_THREAD_LOCAL.remove();if (client ! null) {executorService.execute(() - {try {client.close();} catch (IOException e) {e.printStackTrace();}});}}
}
在使用时同样可以这样
public class MyService {public void someMethod() {RestHighLevelClient client EsClientManager.getClient();try {// 执行操作} finally {EsClientManager.closeClient(client);}}
}
这样每个线程都会从EsClientManager获取一个客户端并在方法结束时自动关闭。
线程池不允许使用 Executors 去创建而是通过 ThreadPoolExecutor 的方式进一步优化
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.LinkedBlockingQueue;public class EsClientManager {private final ExecutorService executorService;private final ThreadLocalRestHighLevelClient CLIENT_THREAD_LOCAL ThreadLocal.withInitial(() - {RestHighLevelClient client null;try {client RestHighLevelClient.builder(new HttpHost(localhost, 9200, http)).build();return client;} catch (IOException e) {e.printStackTrace();return null;}});public EsClientManager() {// 配置 ThreadPoolExecutorint corePoolSize 10; // 核心线程数int maximumPoolSize 10; // 最大线程数long keepAliveTime 0L; // 空闲线程等待新任务的最长时间TimeUnit unit TimeUnit.MILLISECONDS; // keepAliveTime的时间单位int queueCapacity 100; // 工作队列的容量ThreadPoolExecutor executor new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime,unit,new LinkedBlockingQueue(queueCapacity), // 工作队列runnable - {Thread thread new Thread(runnable);thread.setDaemon(false); // 设置线程是否为守护线程false表示非守护线程return thread;});this.executorService Executors.unconfigurableExecutorService(executor);}public RestHighLevelClient getClient() {return CLIENT_THREAD_LOCAL.get();}public void closeClient(RestHighLevelClient client) {CLIENT_THREAD_LOCAL.remove();if (client ! null) {executorService.execute(() - {try {client.close();} catch (IOException e) {e.printStackTrace();}});}}// 添加方法以允许关闭executorServicepublic void shutdown() {executorService.shutdown();}public boolean isShutdown() {return executorService.isShutdown();}public boolean isTerminated() {return executorService.isTerminated();}
}
