Java向ES库中插入数据报错：I/O reactor status: STOPPED

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 ThreadLocal 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 ThreadLocal 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() {
            // 配置 ThreadPoolExecutor
            int 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();
                    }
                });
            }
        }
        // 添加方法以允许关闭executorService
        public void shutdown() {
            executorService.shutdown();
        }
        public boolean isShutdown() {
            return executorService.isShutdown();
        }
        public boolean isTerminated() {
            return executorService.isTerminated();
        }
    }