JobRunr单Worker配置下多任务并发执行问题求助

核心问题分析 #

你配置的worker-count=1确实能保证单节点内任务串行执行，但JobRunr的Worker在一次轮询（由poll-interval-in-seconds控制）中会拉取所有就绪的任务。如果批量提交的任务都是立即执行的，Worker会快速依次执行它们，导致任务启动时间间隔极短（几乎可以忽略），这就是你看到所有任务在同一秒启动的原因。

要实现任务间隔数分钟执行，关键是让任务在不同的时间点就绪，而非依赖Worker的串行执行逻辑。

解决方案1：手动为任务设置递增延迟 #

在批量提交任务时，直接为每个任务设置递增的延迟时间，确保它们在不同时间点被执行。适合业务代码中明确是批量提交的场景。

修改你的测试代码如下：

@Test
void testJobsAreRunSequentially() {
    // Given：设置测试用的任务间隔为2秒（生产环境可改为5分钟）
    Duration jobInterval = Duration.ofSeconds(2);
    for (int i = 0; i < jobs; i++) {
        // 每个任务依次延迟 jobInterval * i 的时间
        BackgroundJob.schedule(jobInterval.multipliedBy(i), () -> jobRunrTestStorageService.add());
    }

    // When
    await()
        .atMost(jobs * jobInterval.getSeconds() + pollInterval, SECONDS)
        .pollInterval(pollInterval, SECONDS)
        .until(() -> jobRunrTestStorageService.getStorage().size() == jobs);

    // Then：验证间隔符合预期
    var jobResults = jobRunrTestStorageService.getStorage();
    for (int i = 1; i < jobs; i++) {
        var timeDiff = ChronoUnit.SECONDS.between(jobResults.get(i - 1), jobResults.get(i));
        assertThat(timeDiff).isGreaterThanOrEqualTo(jobInterval.getSeconds());
    }
}

生产环境中，只需将jobInterval改为Duration.ofMinutes(5)即可。

解决方案2：自定义JobFilter自动添加延迟 #

如果不想在业务代码中硬编码延迟逻辑，可以通过自定义JobFilter，在任务创建阶段自动为批量提交的任务添加递增延迟。

实现JobFilter

@Component
public class BatchJobIntervalFilter implements JobFilter {

    // 线程本地计数器，避免多线程批量提交时互相干扰
    private final ThreadLocal<Integer> batchCounter = ThreadLocal.withInitial(() -> 0);
    private final Duration interval;

    // 从配置文件注入间隔时间（生产环境配置为5分钟）
    public BatchJobIntervalFilter(@Value("${job.batch.interval-minutes:5}") int intervalMinutes) {
        this.interval = Duration.ofMinutes(intervalMinutes);
    }

    @Override
    public void onCreating(JobCreatingContext context) {
        // 为当前线程提交的第N个任务设置延迟
        int count = batchCounter.get();
        context.setScheduledAt(LocalDateTime.now().plus(interval.multipliedBy(count)));
        batchCounter.set(count + 1);
    }

    @Override
    public void onCreated(JobCreatedContext context) {
        // 任务提交完成后重置计数器，确保下一批任务重新开始计数
        batchCounter.set(0);
    }
}

使用方式

业务代码无需修改，保持原有的批量提交逻辑即可：

for (int i = 0; i < jobs; i++) {
    BackgroundJob.enqueue(() -> jobRunrTestStorageService.add());
}

Filter会自动为同一线程中连续提交的任务添加递增延迟，实现间隔执行的效果。

注意事项 #

1. 多节点场景：如果多节点同时提交批量任务，需要额外的批量标记逻辑（比如为任务添加元数据），确保同一批次的任务不会被多个节点重复调度。
2. 轮询间隔配置：poll-interval-in-seconds建议设置为小于任务间隔的值（比如生产环境5分钟任务间隔，轮询间隔设为1分钟），确保Worker能及时发现就绪的任务。

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