JSR352-Wildfly/JBeret：多线程批处理异常时如何暂停并恢复线程？

针对你遇到的数据库性能/内存问题导致作业失败的情况，完全可以实现数据库异常时暂停所有采集线程，清理后恢复运行。结合你的JBeret+Wildfly 10.1.0环境，我整理了一套实用的实现方案：

1. 利用JBeret原生的作业暂停/恢复机制 #

JBeret本身支持作业的suspend（暂停）和resume（恢复）操作，关键是要配置好chunk步骤的检查点，确保恢复时能从最近的进度继续，避免重复处理数据：

• 在你的作业XML配置中，给chunk步骤添加检查点策略和提交间隔：

  <step id="step1" next="step2">
      <chunk checkpoint-policy="item" commit-interval="1000">
          <reader ref="dbDataReader"/>
          <processor ref="dataProcessor"/>
          <writer ref="dataWriter"/>
      </chunk>
      <threads>10</threads>
  </step>
  

  这里checkpoint-policy="item"表示每处理指定数量的条目就保存检查点，commit-interval根据你的数据量调整，平衡性能和恢复粒度。

2. 捕获数据库异常并触发暂停 #

在你的chunk reader/processor中，捕获数据库相关的异常（比如超时、连接失败、锁等待等），判断为数据库性能问题时触发作业暂停：

@Inject
private BatchContext batchContext;

@Override
public Object readItem() throws Exception {
    try {
        // 你的数据库数据读取逻辑
        return fetchDataFromDb();
    } catch (SQLException e) {
        // 根据错误码/异常信息判断是否为数据库性能问题
        if (isDbPerformanceError(e)) {
            final JobOperator jobOperator = BatchRuntime.getJobOperator();
            // 暂停当前作业执行
            jobOperator.suspend(batchContext.getExecutionId());
            // 抛出自定义异常告知JBeret作业已暂停
            throw new JobSuspendException("Detected database performance issue, job suspended");
        }
        // 非目标异常直接抛出
        throw e;
    }
}

// 自定义方法判断数据库异常类型
private boolean isDbPerformanceError(SQLException e) {
    // 示例：匹配超时、连接池耗尽等错误码
    String sqlState = e.getSQLState();
    int errorCode = e.getErrorCode();
    return "HYT00".equals(sqlState) // JDBC超时
            || errorCode == 1205;   // MySQL锁超时
}

3. 调整Wildfly事务与内存配置 #

你遇到的RollbackException是事务超时导致的，需要优化Wildfly的事务和内存配置：

• 事务超时调整：在standalone.xml的transactions子系统延长默认超时时间：

  <subsystem xmlns="urn:jboss:domain:transactions:3.0">
      <coordinator-environment default-timeout="3600"/> <!-- 改为1小时，按需调整 -->
      <!-- 其他配置保持不变 -->
  </subsystem>
  

• 内存配置优化：因为使用内存式job-repository，需确保Wildfly堆内存足够，修改启动参数：

  -Xmx16G -Xms8G -XX:+UseG1GC -XX:MaxGCPauseMillis=200
  

  同时开启GC日志监控内存使用，及时发现内存泄漏或溢出问题。

4. 作业恢复的两种方式 #

当数据库清理完成后，你可以选择：

• 手动恢复：通过Wildfly管理控制台（http://localhost:9990）找到对应的作业执行记录，点击"Resume"按钮恢复。
• 自动恢复：编写一个定时任务（比如用Quartz或Wildfly的定时器），监控数据库状态，当数据库恢复正常后调用JobOperator恢复作业：

  JobOperator jobOperator = BatchRuntime.getJobOperator();
  // 获取所有暂停状态的作业执行
  List<Long> suspendedExecutions = jobOperator.getJobExecutions(jobName)
          .stream()
          .filter(e -> BatchStatus.SUSPENDED.equals(jobOperator.getJobExecution(e).getBatchStatus()))
          .collect(Collectors.toList());
  // 恢复作业
  suspendedExecutions.forEach(jobOperator::resume);
  

5. 额外优化建议 #

• 调整数据库连接池大小：在Wildfly数据源配置中增大max-pool-size（比如设为20），避免10个采集线程耗尽连接。
• 拆分作业：将5天的海量数据按时间/ID拆分多个子作业，降低单个作业的压力和恢复风险。
• 实时监控：通过Wildfly的监控工具或JBeret的metrics，跟踪作业的线程、内存、数据库连接使用情况，提前预警问题。

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