Elasticsearch Java客户端无法创建本地线程OOM问题求助

首先明确：你遇到的java.lang.OutOfMemoryError: unable to create new native thread和Java堆内存无关——堆内存不足的错误提示是OutOfMemoryError: Java heap space，而这个错误是操作系统无法为Java进程创建更多原生线程导致的，结合你的代码和配置来看，核心原因和解决办法如下：

一、错误根源定位 #

1. 代码中的致命问题：重复创建TransportClient #

你的代码里多次调用new ConfigureES().client：

• 在BulkProcessor.builder里新建了一个客户端实例
• 在catch块里又新建了一个客户端实例

每个Elasticsearch TransportClient内部会初始化一个ThreadPool，包含多个线程池（比如bulk、index、search等），每个线程池会创建若干原生线程。重复创建客户端会快速耗尽系统的线程资源，这是导致错误的直接原因。

2. 对比Python客户端的差异 #

Python的Elasticsearch客户端基于HTTP协议，使用连接池管理请求，不会像Java TransportClient那样创建大量原生线程，所以能正常运行，这也侧面验证了问题出在Java客户端的线程创建逻辑上。

3. 系统线程配置的辅助验证 #

你的ulimit -a显示最大用户进程数（-u）：128065，这个上限足够，但Java进程的线程数已经达到208，重复创建客户端会让这个数值持续增长，直到触发系统的线程创建限制。

二、具体解决办法 #

1. 修复代码：复用唯一的TransportClient实例 #

修改你的批量导入代码，只创建一次客户端并复用，避免重复初始化：

public ReadFromCsvAndImportToEs() throws IOException, NoSuchAlgorithmException, ParseException { 
    // 仅初始化一次Elasticsearch客户端
    TransportClient client = new ConfigureES().client;
    BulkProcessor bulkProcessor = null;

    try {
        // 基于同一个客户端构建BulkProcessor
        bulkProcessor = BulkProcessor.builder( 
            client, 
            new BulkProcessor.Listener() { 
                @Override public void beforeBulk(long executionId,BulkRequest request) {} 
                @Override public void afterBulk(long executionId,BulkRequest request,BulkResponse response) {} 
                @Override public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                    // 建议添加失败日志，方便排查问题
                    System.err.println("Bulk execution failed: " + failure.getMessage());
                    failure.printStackTrace();
                } 
            }) 
            .setBulkActions(10000) 
            .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB)) 
            .setFlushInterval(TimeValue.timeValueSeconds(5)) 
            .setConcurrentRequests(1) 
            .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3)) 
            .build(); 

        BufferedReader br = new BufferedReader(new FileReader("/data/months/modified/nov-17-dec-17.csv")); 
        br.readLine(); // 跳过CSV表头
        String line; 
        String[] lines; 

        while((line = br.readLine())!=null){ 
            lines = line.split(","); 
            // 字段处理逻辑保持不变
            String store_code = lines[3]; 
            String bill_date = lines[22]; 
            String cart_id = md5(lines[20]).substring(0,15).toUpperCase(); 
            String bill_no = md5(lines[22]+"-"+lines[3]).substring(1,18).toUpperCase(); 
            String division = lines[21]; 
            String icode = md5(lines[33]).substring(0,5).toUpperCase(); 
            String mrp = lines[25]; 
            String qty = lines[27]; 
            String totalAmt = lines[35]; 
            String section = lines[2]; 
            String department = lines[19]; 

            bulkProcessor.add( 
                new IndexRequest("index", "type") 
                    .source(jsonBuilder() 
                        .startObject() 
                        .field("store_code", store_code) 
                        .field("bill_date", bill_date) 
                        .field("bill_no", bill_no) 
                        .field("cart_id", cart_id) 
                        .field("division", division) 
                        .field("icode", icode) 
                        .field("mrp", Double.parseDouble(mrp)) 
                        .field("qty", Double.parseDouble(qty)) 
                        .field("totalAmt", Double.parseDouble(totalAmt)) 
                        .field("section", section) 
                        .field("department", department) 
                        .endObject() 
                    ) 
            ); 
        } 

        // 等待所有批量请求处理完成再关闭
        if (bulkProcessor != null) {
            bulkProcessor.close();
            bulkProcessor.awaitClose(10, TimeUnit.MINUTES);
        }
    } catch (Exception e) { 
        System.err.println("Error during import: " + e.getMessage());
        e.printStackTrace();
    } finally {
        // 最后统一关闭唯一的客户端实例
        if (client != null) {
            client.close();
        }
    }
}

2. 优化系统线程限制（可选） #

如果修复代码后仍有问题，可以调整系统的线程相关配置：

• 临时调整当前会话的线程限制：

  # 减小线程栈大小（让相同内存能创建更多线程）
  ulimit -s 10240
  # 增大用户进程/线程上限
  ulimit -u 65535
  

• 永久调整（修改/etc/security/limits.conf）：

  * soft nproc 65535
  * hard nproc 65535
  * soft stack 10240
  * hard stack 10240
  

  修改后重启会话或机器生效。

3. 优化Elasticsearch客户端线程池 #

在ConfigureES类中创建客户端时，自定义线程池大小，减少线程占用：

Settings settings = Settings.builder()
    .put("threadpool.bulk.size", 2) // 减小bulk线程池大小，默认是CPU核心数
    .put("threadpool.index.size", 2) // 减小index线程池大小
    .build();

TransportClient client = new PreBuiltXPackTransportClient(settings)
    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("你的ES地址"), 9300));

三、验证方法 #

运行修改后的代码，同时监控Java进程的线程数：

# 先获取Java进程PID
ps aux | grep java
# 实时监控线程数
watch -n 1 'ps -eLF | grep <你的Java进程PID> | wc -l'

如果线程数稳定在几十到一百左右，不再持续暴涨，说明问题已解决。

内容的提问来源于stack exchange，提问作者Saurab