使用cron在Spring中安排周期性的反应任务？答案

【问题标题】：Scheduling periodic reactive tasks in Spring using cron?使用cron在Spring中安排周期性的反应任务？
【发布时间】：2019-05-19 00:49:48
【问题描述】：

通常我会做这样的事情来安排一个作业在 Spring 中定期执行，并在给定的时区使用 cron：

@Scheduled(cron = "0 0 10 * * *", zone = "Europe/Stockholm")
public void scheduleStuff() {
    // Do stuff
}

这将阻塞调用scheduleStuff 的线程，直到作业完成。但是在这种情况下，我想做的“东西”都是使用 Springs 的project reactor 的非阻塞构建块（即Mono、Flux 等）实现的。

例如假设我想定期触发这个功能：

Flux<Void> stuff() {
    return ..
}

我当然可以简单地调用stuff().subscribe()（甚至stuff().block()），但这会阻塞线程。对于非阻塞代码，有没有更好的方法来实现与@Scheduled(cron = "0 0 10 * * *", zone = "Europe/Stockholm") 相同的功能？

我使用的是 Spring Boot 2.1。

【问题讨论】：

实际上，subscribe() 不会阻止您的线程。如果你真的需要，你可以调用stuff().subscribeOn(Schedulers.parallel()).subscribe() 或其他调度程序来确保执行将在一个单独的线程中完成。
@AlexanderPankin 我相信你可以这样回答

标签： spring spring-boot cron spring-webflux project-reactor

【解决方案1】：

实际上，subscribe() 不会阻塞您的线程。如果您确实需要，您可以调用stuff().subscribeOn(Schedulers.parallel()).subscribe() 或其他调度程序以确保执行将在单独的线程中完成。

【讨论】：

【解决方案2】：

您可以将 stuff 方法包装在异步方法中

例如：

@Scheduled(cron = "0 0 10 * * *", zone = "Europe/Stockholm")
public void scheduleStuff() {
    stuffService.doStuffAsync();
}

使用异步方法的服务

public class StuffService() implements IStuffService {

    @Async
    public void doStuffAsync() {
       // Call and subscribe to your flux method here
    }

}

对doStuffAsync() 的调用将立即返回scheduleStuff，因此不会阻塞线程。

【讨论】：

@BrianClozel 我没有遵循你的逻辑。 doStuffAsync() 将调用并订阅 stuff() 方法。这与接受的答案建议的相同。

【解决方案3】：

这里还有一个选项：

public class PeriodicReactiveTasksInSpring implements SmartLifecycle {

    private final AtomicReference<Subscription> subscription;
    private final Long executionPeriod;

    public PeriodicReactiveTasksInSpring(Long executionPeriod) {
        this.subscription = new AtomicReference<>();
        this.executionPeriod = executionPeriod;
    }

    @Override
    public void start() {
        if (Objects.isNull(subscription.get())) {
            updateConfig()
                    .doOnSubscribe(sub -> {
                        subscription.set(sub);
                    }).subscribe();
        }
    }

    @Override
    public void stop() {
        Optional.ofNullable(subscription.get())
                .ifPresent(sub -> {
                    sub.cancel();
                    subscription.set(null);
                });
    }

    @Override
    public boolean isRunning() {
        return Objects.nonNull(subscription.get());
    }


    private Flux<Item> updateConfig() {
        return Flux.interval(Duration.ofMillis(executionPeriod))
                .subscribeOn(Schedulers.boundedElastic())
                .flatMap(cfg -> {
                    // Do your job here
                })
                .onErrorContinue((err, msg) -> LOGGER.error("Error: {} message: {}", err, msg));
    }

}

【讨论】：