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

Posted 2023-02-27

【中文标题】使用cron在Spring中安排周期性的反应任务？

【英文标题】：Scheduling periodic reactive tasks in Spring using cron?

【发布时间】：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 我相信你可以这样回答

【参考方案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));