Scheduling periodic reactive tasks in Spring using cron?

user_398441 0
Viewed 6127

Normally I would do something like this to schedule a job to be periodically executed in Spring with cron in a given timezone:

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

This is will block the thread calling scheduleStuff until the job is completed. However in this case the "stuff" I want to do is all implemented using Springs' non-blocking building blocks of project reactor (i.e. Mono, Flux etc).

E.g. let's say that I want to trigger this function periodically:

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

I can of course simply call stuff().subscribe() (or even stuff().block()) but this will block the thread. Is there a better way to achieve the same things as @Scheduled(cron = "0 0 10 * * *", zone = "Europe/Stockholm") for non-blocking code?

I'm using Spring Boot 2.1.

project-reactor spring-webflux spring-boot cron spring
3 Answers

Actualy, subscribe() doesn't block your thread. You could call stuff().subscribeOn(Schedulers.parallel()).subscribe() or other scheduler to be sure that execution will be done in a separate thread, if you really need it.

user_102644300

You could wrap the stuff method in an async method

Ex:

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

Service with async method

public class StuffService() implements IStuffService {

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

}

The call to doStuffAsync() will return immediately the the scheduleStufftherefor not blocking the thread.

user_32252600

Here is one more option:

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));
    }

}
user_40941710
Related