WebLogic连接池资源耗尽：并行流批量处理优化方案咨询

问题描述 #

从Oracle数据库获取1000条记录后，使用Java并行流进行并发处理时，抛出以下错误导致任务中断：

weblogic.common.resourcepool.ResourceLimitException: No resources currently available in pool ERDS to allocate to applications, please increase the size of the pool and retry

改用串行流可正常执行，但面对2万条数据时仍需多线程处理，且因业务限制无法使用PL/SQL实现逻辑，需在Java层面优化现有代码。

根本原因 #

1. 并行流无限制并发：默认ForkJoinPool的线程数基于CPU核心数，每个并行线程会调用EJB方法，而每个EJB调用会占用一个数据库连接，瞬间耗尽连接池资源。
2. 频繁数据库交互：joinWithdebit方法内多次执行独立查询，每个查询都占用连接，进一步加剧资源消耗。
3. 事务管理不当：sendSms方法使用REQUIRES_NEW事务属性，每次调用都会开启新事务，延长连接占用时间。

优化方案 #

1. 控制并发线程数 #

避免并行流无限制占用连接，改用自定义线程池或指定并行流的并发数，确保并发数不超过连接池可用资源的合理比例（建议为连接池大小的70%）。

修改TPTimer.sendSMS方法：

private void sendSMS() {
    try {
        String txt = ".....";
        List<Object[]> list = em.createNativeQuery("some query")
                .setMaxResults(1000)
                .getResultList();

        // 自定义固定大小线程池，根据连接池容量调整线程数
        ExecutorService executor = Executors.newFixedThreadPool(10);
        for (Object[] item : list) {
            final Object[] arr = item;
            executor.submit(() -> service.joinWithdebit(txt, arr));
        }
        executor.shutdown();
        // 等待所有任务完成，超时时间按需调整
        executor.awaitTermination(2, TimeUnit.HOURS);
    } catch (Exception e) {
        logger.error("短信发送任务失败", e);
    }
}

或者使用自定义ForkJoinPool控制并行流并发：

private void sendSMS() {
    try {
        String txt = ".....";
        List<Object[]> list = em.createNativeQuery("some query")
                .setMaxResults(1000)
                .getResultList();

        // 自定义ForkJoinPool，指定并发数
        ForkJoinPool customPool = new ForkJoinPool(10);
        customPool.submit(() -> list.parallelStream().forEach(t -> service.joinWithdebit(txt, t))).get();
        customPool.shutdown();
    } catch (Exception e) {
        logger.error("短信发送任务失败", e);
    }
}

2. 合并数据库查询 #

将joinWithdebit方法内的两次独立查询合并为一个SQL语句，减少数据库连接的占用次数和交互开销。

修改Service.joinWithdebit方法：

public void joinWithdebit(String txt, Object[] arr) {
    try {
        // 合并两次查询为一个JOIN语句，根据实际业务逻辑调整关联条件
        String mergedQuery = "SELECT 1 FROM " +
                            "(SELECT * FROM table1 WHERE col = ?) q1 " +
                            "JOIN " +
                            "(SELECT * FROM table2 WHERE col1 = ? AND col2 = ?) q2 " +
                            "ON q1.id = q2.ref_id"; // 替换为实际关联逻辑

        List<?> res = em.createNativeQuery(mergedQuery)
                .setParameter(1, String.valueOf(arr[2]))
                .setParameter(2, String.valueOf(arr[1]))
                .setParameter(3, String.valueOf(arr[0]))
                .getResultList();

        if (!res.isEmpty()) {
            sendSms(txt, arr);
        }
    } catch (Exception e) {
        Logger.getLogger(Service.class.getName()).error("查询验证失败", e);
    }
}

3. 优化事务与连接使用 #

• 缩短事务执行时间：在sendSms方法中，尽量将非数据库操作（如日期格式化）移到事务外。
• 避免不必要的joinTransaction：EJB容器会自动管理事务，除非手动控制事务，否则无需显式调用em.joinTransaction()。

优化Service.sendSms方法：

@TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
public void sendSms(String txt, Object[] arr) {
    try {
        // 预格式化日期，移到事务外
        String now = DateUtils.format(new Date(), "yyyyMMdd");
        String userName = String.valueOf(arr[0]);
        
        User user = userManager.getUserByName(userName);
        boolean bool = smsService.isSmsSent(user, txt);
        
        if (bool) {
            em.createNativeQuery("insert into sms_table(col1,col2,col3,col4,col5,col6,col7,col8,create_time) values (?,?,?,?,?,?,?,?,SYSDATE)")
                    .setParameter(1, new BigDecimal(userName + "5" + "01"))
                    .setParameter(2, String.valueOf(arr[2]))
                    .setParameter(3, "01")
                    .setParameter(4, userName)
                    .setParameter(5, now)
                    .setParameter(6, now)
                    .setParameter(7, user != null ? user.getMobile() : "-")
                    .setParameter(8, "5")
                    .executeUpdate();
        }
    } catch (Exception e) {
        Logger.getLogger(Service.class.getName()).error("320220:短信发送失败", e);
    }
}

4. 批量处理优化（可选） #

对于2万条数据，可将数据分批处理，每批处理500条左右，避免一次性加载大量数据到内存，同时控制每批的并发数，进一步降低连接池压力。

总结 #

通过控制并发线程数、合并数据库查询、优化事务管理，既能保持多线程处理的性能优势，又能避免连接池资源耗尽的问题。调整线程数时需结合WebLogic连接池的配置（如初始连接数、最大连接数），确保并发数在合理范围内。

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