基于Spring WebFlux的Job执行与管理服务架构设计咨询

问题背景 #

• 接收多客户端提交的独立Job，每个Job包含按顺序执行的Task链，Task依赖前序执行结果
• 单个Task耗时差异极大（数分钟至数小时），Job最长可耗时1-2天
• 客户端需支持：随时取消Job、动态修改Job（移除Task、中断当前Task跳转至下一个）、实时查询Job进度
• 计划基于Spring WebFlux实现，初步考虑Master-Worker模式，但该模式Worker通常处理同一Job的分片，与当前场景（Worker处理独立Job）不匹配，需适配方案

适配设计方案：JobManager + 独立JobExecutor模式 #

核心角色划分 #

• JobManager：全局管控节点，负责Job的接收、持久化、状态维护、响应客户端的查询/修改/取消请求、JobExecutor的调度与监控
• JobExecutor：每个实例对应一个独立Job的完整生命周期，负责按顺序执行Task链、处理Job的动态修改/取消指令、实时上报Task/Job状态至JobManager

模式优势（对比传统Master-Worker） #

• 每个JobExecutor完全独立，专注于单个Job的顺序Task执行，完美适配多独立Job的场景
• JobManager仅做管控，不参与具体执行，避免单点性能瓶颈
• 天然支持多Job并行处理，各Job生命周期完全隔离

基于Spring WebFlux的关键实现细节 #

1. Job/Task状态持久化

采用Reactive数据库（如R2DBC、MongoDB Reactive）存储Job和Task的状态、执行上下文、前序Task结果，确保服务重启后可恢复执行：

• Job状态：待执行、执行中、已暂停、已取消、已完成、失败
• Task状态：待执行、执行中、已完成、已中断、失败
• 持久化前序Task的输出结果，供后续Task读取依赖

2. JobExecutor的动态管控

每个JobExecutor对应一条Reactive Stream（如Flux），驱动Task的顺序执行：

• 启动Job时，JobManager创建对应JobExecutor实例，生成Task执行的Flux链，每个Task的执行依赖前序Mono的结果
• 客户端发起取消/修改请求时，JobManager向对应JobExecutor发送信号（如CancellationSignal或自定义事件），JobExecutor通过Disposable或Flux#takeUntilOther中断当前Task，并根据修改后的Task链重新生成执行流

示例伪代码：

// JobExecutor核心执行逻辑
Flux.fromIterable(job.getTaskList())
    .concatMap(task -> Mono.defer(() -> {
        // 检查中断/修改信号
        if (isCancelled()) {
            return Mono.error(new JobCancelledException());
        }
        if (isTaskSkipped(task)) {
            return Mono.empty(); // 跳过当前Task，执行下一个
        }
        // 执行Task，传入前序结果
        return taskExecutor.execute(task, previousTaskResult)
            .doOnSuccess(result -> {
                // 上报Task完成状态至JobManager
                jobManager.reportTaskStatus(job.getId(), task.getId(), TaskStatus.COMPLETED, result);
            })
            .doOnError(e -> {
                jobManager.reportTaskStatus(job.getId(), task.getId(), TaskStatus.FAILED, e.getMessage());
            });
    }))
    .doOnComplete(() -> jobManager.reportJobStatus(job.getId(), JobStatus.COMPLETED))
    .doOnError(e -> jobManager.reportJobStatus(job.getId(), JobStatus.FAILED, e.getMessage()))
    .subscribe();

3. 实时进度查询

JobManager通过Reactive接口暴露查询端点，直接从Reactive数据库读取最新的Job/Task状态返回给客户端：

@GetMapping("/jobs/{jobId}/status")
public Mono<JobStatusResponse> getJobStatus(@PathVariable String jobId) {
    return jobRepository.findById(jobId)
        .map(job -> {
            long completedCount = job.getTasks().stream()
                .filter(t -> t.getStatus() == TaskStatus.COMPLETED)
                .count();
            return JobStatusResponse.builder()
                .jobId(jobId)
                .jobStatus(job.getStatus())
                .completedTasks(completedCount)
                .totalTasks(job.getTasks().size())
                .build();
        });
}

4. Job动态修改处理

当客户端提交修改请求（如移除Task、跳转Task），JobManager执行以下步骤：

1. 更新数据库中的Job配置
2. 向对应JobExecutor发送修改事件
3. JobExecutor收到事件后，中断当前执行的Task（若正在执行），从修改后的Task链指定位置继续执行

扩展与优化思路 #

• 资源隔离：利用Reactor的Schedulers为不同JobExecutor分配独立线程池，避免长时Task占用过多资源影响其他Job
• Task可中断性：要求所有Task实现中断逻辑（如定期检查中断标志、响应Disposable取消信号），确保Task可被强制停止
• 暂停/恢复功能：在JobExecutor中支持暂停信号，暂停时持久化当前执行上下文，恢复时从暂停的Task继续执行
• 集群部署：将JobManager做成集群，用分布式锁保证Job的唯一调度；JobExecutor可部署在多节点，通过服务发现注册，JobManager根据负载分配Job

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