如何在Java Flux中实现递增延迟的循环执行逻辑？

核心思路 #

把循环拆分为三个明确的阶段，通过Flux.concat按顺序执行每个阶段，每个阶段用interval控制执行间隔，用takeUntil控制阶段时长：

• 阶段1：前5分钟，每10秒执行一次
• 阶段2：接下来10分钟，每30秒执行一次
• 阶段3：永久循环，每1分钟执行一次

完整代码实现 #

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;

public class DelayedLoopExample {
    public static void main(String[] args) {
        // 定义各阶段的延迟和持续时长
        Duration phase1Delay = Duration.ofSeconds(10);
        Duration phase1Total = Duration.ofMinutes(5);
        Duration phase2Delay = Duration.ofSeconds(30);
        Duration phase2Total = Duration.ofMinutes(10);
        Duration phase3Delay = Duration.ofMinutes(1);

        // 第一阶段：前5分钟按10秒间隔执行
        Flux<Void> phase1 = Flux.interval(phase1Delay)
                .takeUntil(tick -> tick * phase1Delay.toMillis() >= phase1Total.toMillis())
                .flatMap(tick -> runBusinessLogic());

        // 第二阶段：接下来10分钟按30秒间隔执行
        Flux<Void> phase2 = Flux.interval(phase2Delay)
                .takeUntil(tick -> tick * phase2Delay.toMillis() >= phase2Total.toMillis())
                .flatMap(tick -> runBusinessLogic());

        // 第三阶段：永久按1分钟间隔执行
        Flux<Void> phase3 = Flux.interval(phase3Delay)
                .flatMap(tick -> runBusinessLogic());

        // 拼接三个阶段并启动执行
        Flux.concat(phase1, phase2, phase3)
                .subscribeOn(Schedulers.boundedElastic())
                .subscribe(
                        null,
                        err -> System.err.println("业务执行出错: " + err.getMessage()),
                        () -> System.out.println("循环执行结束")
                );

        // 阻塞主线程防止程序退出
        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // 模拟实际业务逻辑，替换为你的代码即可
    private static Flux<Void> runBusinessLogic() {
        return Flux.fromRunnable(() -> {
            System.out.printf("执行业务逻辑 - 时间戳: %d%n", System.currentTimeMillis());
            // 这里写你的业务代码
        }).then();
    }
}

关键代码解释 #

• 阶段拼接：Flux.concat确保三个阶段按顺序执行，前一阶段完全结束后才进入下一阶段，避免间隔逻辑混乱。
• 阶段时长控制：takeUntil通过计算已运行时间（tick * 间隔时长）判断是否达到阶段总时长，精准控制每个阶段的运行时间。
• 业务逻辑封装：runBusinessLogic返回Flux<Void>，适配Reactor的异步模型，不管业务是同步还是异步都能无缝集成。
• 线程调度：subscribeOn(Schedulers.boundedElastic())将业务逻辑放到弹性线程池中执行，避免阻塞主线程，适合IO密集型任务。

补充说明 #

如果业务逻辑本身是异步的，可以直接返回对应的Flux/Mono，不需要用Flux.fromRunnable包裹；如果业务是同步且耗时较长，建议在runBusinessLogic内部用subscribeOn指定单独的线程池，避免影响其他阶段的调度。

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