Spring Batch多实例远程分区：无消息中间件如何触发实例执行？

你已经完成了记录与实例的分配，以下是几种不需要消息中间件的触发方案：

方案一：数据库触发信号机制 #

利用已有的数据库资源，通过一个全局触发表来统一通知所有实例启动任务：

• 创建触发表，用于存储触发状态：

CREATE TABLE batch_trigger (
    id INT PRIMARY KEY AUTO_INCREMENT,
    trigger_status VARCHAR(20) DEFAULT 'IDLE', -- 可选值：IDLE/PENDING/EXECUTING/COMPLETED
    trigger_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

• 每个Spring Batch实例启动后，添加定时轮询逻辑（用Spring的@Scheduled），定期查询触发表状态：

@Component
public class BatchTriggerListener {
    private final JdbcTemplate jdbcTemplate;
    private final JobLauncher jobLauncher;
    private final Job partitionedJob;
    private final String instanceId; // 实例标识，从配置文件读取

    public BatchTriggerListener(JdbcTemplate jdbcTemplate, JobLauncher jobLauncher, Job partitionedJob, @Value("${instance.id}") String instanceId) {
        this.jdbcTemplate = jdbcTemplate;
        this.jobLauncher = jobLauncher;
        this.partitionedJob = partitionedJob;
        this.instanceId = instanceId;
    }

    @Scheduled(fixedDelay = 5000) // 每5秒轮询一次
    public void checkTriggerStatus() throws Exception {
        String status = jdbcTemplate.queryForObject("SELECT trigger_status FROM batch_trigger ORDER BY id DESC LIMIT 1", String.class);
        if ("PENDING".equals(status)) {
            // 抢占执行状态，避免多个实例重复处理
            int updateCount = jdbcTemplate.update("UPDATE batch_trigger SET trigger_status = 'EXECUTING' WHERE trigger_status = 'PENDING'");
            if (updateCount > 0) {
                // 启动当前实例的任务，仅处理分配给自己的记录
                JobParameters params = new JobParametersBuilder()
                        .addString("instanceId", instanceId)
                        .addLong("timestamp", System.currentTimeMillis())
                        .toJobParameters();
                jobLauncher.run(partitionedJob, params);
                // 任务完成后更新状态为COMPLETED
                jdbcTemplate.update("UPDATE batch_trigger SET trigger_status = 'COMPLETED' WHERE trigger_status = 'EXECUTING'");
            }
        }
    }
}

• 触发任务时，只需执行SQL更新触发表状态为PENDING，所有实例轮询到后会自动启动任务。

方案二：HTTP接口批量触发 #

给每个实例暴露触发接口，通过脚本或小服务批量调用：

• 每个实例中编写触发接口：

@RestController
@RequestMapping("/batch")
public class BatchTriggerController {
    private final JobLauncher jobLauncher;
    private final Job partitionedJob;
    private final String instanceId;

    public BatchTriggerController(JobLauncher jobLauncher, Job partitionedJob, @Value("${instance.id}") String instanceId) {
        this.jobLauncher = jobLauncher;
        this.partitionedJob = partitionedJob;
        this.instanceId = instanceId;
    }

    @PostMapping("/trigger")
    public ResponseEntity<String> triggerJob() throws Exception {
        JobParameters params = new JobParametersBuilder()
                .addString("instanceId", instanceId)
                .addLong("timestamp", System.currentTimeMillis())
                .toJobParameters();
        jobLauncher.run(partitionedJob, params);
        return ResponseEntity.ok("Instance " + instanceId + " job triggered");
    }
}

• 维护所有实例的地址列表，编写Shell脚本批量调用接口：

#!/bin/bash
INSTANCES=("http://instance1:8080" "http://instance2:8080" "http://instance3:8080")

for INSTANCE in "${INSTANCES[@]}"
do
  curl -X POST "$INSTANCE/batch/trigger"
  echo "Triggered $INSTANCE"
done

• 执行脚本即可批量触发所有实例的任务。

方案三：Spring Cloud Task分布式触发（若已用Spring Cloud） #

若你的架构基于Spring Cloud，可借助Spring Cloud Task的TaskLauncher实现批量触发：

• 每个实例引入Spring Cloud Task依赖，并配置任务属性：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-task</artifactId>
</dependency>

• 创建任务启动服务，通过TaskLauncher批量启动各实例的Batch任务：

@Service
public class BatchTaskLauncher {
    private final TaskLauncher taskLauncher;

    public BatchTaskLauncher(TaskLauncher taskLauncher) {
        this.taskLauncher = taskLauncher;
    }

    public void triggerAllInstances() {
        List<String> instanceUris = Arrays.asList("http://instance1:8080", "http://instance2:8080");
        for (String uri : instanceUris) {
            TaskLaunchRequest request = new TaskLaunchRequest();
            request.setUri(uri);
            request.setArguments(Collections.singletonList("--instance.id=" + uri.substring(uri.lastIndexOf("/")+1)));
            taskLauncher.launch(request);
        }
    }
}

• 调用triggerAllInstances()方法即可触发所有实例执行任务。

注意事项 #

• 每个实例的任务Reader要严格根据instanceId过滤出分配给自己的记录，确保任务执行范围正确。
• 所有方案需考虑幂等性，比如通过JobParameters中的唯一标识（如timestamp）避免重复执行同一任务。

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