如何用Flux Reactor替代Java Scheduler实现每小时定时调度？

我来帮你理清这两个困惑点，用Reactor优雅实现每小时定时调度的效果：

一、避免生成不必要的Long值 #

你之所以会得到多余的Long值，是因为Flux.interval()默认会发射从0开始的递增Long序列（用来标记第几次调度）。但如果我们只需要执行定时任务、不需要这个计数的话，直接忽略它就好。

正确的实现方式 #

根据你的任务类型（同步/异步），可以选下面两种写法：

1. 同步任务场景

如果你的定时任务是同步执行的（比如本地计算、文件操作），用doOnNext()忽略掉发射的Long值，直接执行任务逻辑：

Flux.interval(Duration.ofHours(1))
    .doOnNext(ignored -> {
        // 这里写你的每小时任务逻辑
        System.out.println("执行每小时定时任务：" + LocalDateTime.now());
    })
    .subscribeOn(Schedulers.boundedElastic()) // 指定任务执行的线程池
    .subscribe();

2. 异步任务场景

如果任务是异步的（比如数据库查询、HTTP请求），用flatMap()包裹任务，同样忽略Long参数：

Flux.interval(Duration.ofHours(1))
    .onBackpressureDrop() // 防止任务堆积：如果上一次任务没完成，跳过本次调度
    .flatMap(ignored -> executeAsyncHourlyTask()) // executeAsyncHourlyTask()返回Mono<Void>
    .subscribeOn(Schedulers.boundedElastic())
    .subscribe();

额外需求：立即执行第一次任务

如果希望程序启动后马上执行一次任务，之后再每小时执行，把interval的初始延迟设为Duration.ZERO即可：

Flux.interval(Duration.ZERO, Duration.ofHours(1))
    // ... 后续逻辑同上

二、如何处理Disposable对象 #

subscribe()返回的Disposable是调度任务的生命周期控制器，它的核心作用就是让你能主动停止定时调度。

1. 基础用法：手动控制停止 #

保存Disposable的引用，在需要停止调度的时候（比如用户触发停止、应用关闭前）调用dispose()：

// 全局保存Disposable引用
private Disposable hourlyTaskDisposable;

// 初始化调度
public void startHourlyTask() {
    hourlyTaskDisposable = Flux.interval(Duration.ofHours(1))
        .doOnNext(ignored -> executeHourlyTask())
        .subscribeOn(Schedulers.boundedElastic())
        .subscribe();
}

// 停止调度
public void stopHourlyTask() {
    if (hourlyTaskDisposable != null && !hourlyTaskDisposable.isDisposed()) {
        hourlyTaskDisposable.dispose();
    }
}

2. Spring Boot环境下的优雅处理 #

如果是在Spring Boot项目中，可以把Disposable注册为Bean，并指定destroyMethod = "dispose"，这样Spring容器关闭时会自动调用dispose()，优雅停止调度：

@Configuration
public class HourlyTaskConfig {

    @Bean(destroyMethod = "dispose")
    public Disposable hourlyTaskScheduler() {
        return Flux.interval(Duration.ofHours(1))
            .doOnNext(ignored -> {
                // 任务逻辑
                System.out.println("Spring环境下执行每小时任务");
            })
            .subscribeOn(Schedulers.boundedElastic())
            .subscribe();
    }
}

额外注意事项 #

• 选择合适的Scheduler：Schedulers.boundedElastic()适合IO密集型任务；如果是CPU密集型任务，改用Schedulers.parallel()。
• 背压处理：如果任务执行时间可能超过1小时，一定要加背压策略（比如onBackpressureDrop()），避免任务堆积导致内存溢出。

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