reactor.core.scheduler

Class Schedulers

java.lang.Object

reactor.core.scheduler.Schedulers

public abstract class Schedulers
extends Object

Schedulers provides various Scheduler factories useable by publishOn or subscribeOn :

• parallel(): Optimized for fast Runnable non-blocking executions
• single(): Optimized for low-latency Runnable one-off executions
• elastic(): Optimized for longer executions, an alternative for blocking tasks where the number of active tasks (and threads) can grow indefinitely
• immediate(): to immediately run submitted Runnable instead of scheduling them (somewhat of a no-op or "null object" Scheduler)
• fromExecutorService(ExecutorService) to create new instances around Executors

Factories prefixed with new return a new instance of their flavor of Scheduler, while other factories like elastic() return a shared instance, that is the one used by operators requiring that flavor as their default Scheduler.

Author:

Stephane Maldini

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	Schedulers.Factory Public factory hook to override Schedulers behavior globally

Field Summary

Fields

Modifier and Type	Field and Description
static int	DEFAULT_POOL_SIZE Default pool size, initialized by system property `reactor.schedulers.defaultPoolSize` and falls back to the number of processors available to the runtime on init.

Constructor Summary

Constructors

Constructor and Description
Schedulers()

Method Summary

Modifier and Type	Method and Description
static boolean	addExecutorServiceDecorator(String key, BiFunction<Scheduler,ScheduledExecutorService,ScheduledExecutorService> decorator) Set up an additional ScheduledExecutorService decorator for a given key only if that key is not already present.
static ScheduledExecutorService	decorateExecutorService(Scheduler owner, ScheduledExecutorService original) This method is aimed at Scheduler implementors, enabling custom implementations that are backed by a ScheduledExecutorService to also have said executors decorated (ie.
static void	disableMetrics() If enableMetrics() has been previously called, removes the decorator.
static Scheduler	elastic() Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.
static void	enableMetrics() If Micrometer is available, set-up a decorator that will instrument any ExecutorService that backs a Scheduler.
static Scheduler	fromExecutor(Executor executor) Create a Scheduler which uses a backing Executor to schedule Runnables for async operators.
static Scheduler	fromExecutor(Executor executor, boolean trampoline) Create a Scheduler which uses a backing Executor to schedule Runnables for async operators.
static Scheduler	fromExecutorService(ExecutorService executorService) Create a Scheduler which uses a backing ExecutorService to schedule Runnables for async operators.
static Scheduler	fromExecutorService(ExecutorService executorService, String executorName) Create a Scheduler which uses a backing ExecutorService to schedule Runnables for async operators.
static Scheduler	immediate() Executes tasks immediately instead of scheduling them.
static boolean	isInNonBlockingThread() Check if calling a Reactor blocking API in the current Thread is forbidden or not, by checking if the thread implements NonBlocking (in which case it is forbidden and this method returns true).
static boolean	isNonBlockingThread(Thread t) Check if calling a Reactor blocking API in the given Thread is forbidden or not, by checking if the thread implements NonBlocking (in which case it is forbidden and this method returns true).
static Scheduler	newElastic(int ttlSeconds, ThreadFactory threadFactory) Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.
static Scheduler	newElastic(String name) Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.
static Scheduler	newElastic(String name, int ttlSeconds) Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.
static Scheduler	newElastic(String name, int ttlSeconds, boolean daemon) Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.
static Scheduler	newParallel(int parallelism, ThreadFactory threadFactory) Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work.
static Scheduler	newParallel(String name) Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work.
static Scheduler	newParallel(String name, int parallelism) Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work.
static Scheduler	newParallel(String name, int parallelism, boolean daemon) Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work.
static Scheduler	newSingle(String name) Scheduler that hosts a single-threaded ExecutorService-based worker and is suited for parallel work.
static Scheduler	newSingle(String name, boolean daemon) Scheduler that hosts a single-threaded ExecutorService-based worker and is suited for parallel work.
static Scheduler	newSingle(ThreadFactory threadFactory) Scheduler that hosts a single-threaded ExecutorService-based worker and is suited for parallel work.
static void	onHandleError(BiConsumer<Thread,? super Throwable> c) Define a hook that is executed when a Scheduler has handled an error.
static Scheduler	parallel() Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work.
static BiFunction<Scheduler,ScheduledExecutorService,ScheduledExecutorService>	removeExecutorServiceDecorator(String key) Remove an existing ScheduledExecutorService decorator if it has been set up via addExecutorServiceDecorator(String, BiFunction).
static void	resetFactory() Re-apply default factory to Schedulers
static void	resetOnHandleError() Reset the onHandleError(BiConsumer) hook to the default no-op behavior.
static void	setExecutorServiceDecorator(String key, BiFunction<Scheduler,ScheduledExecutorService,ScheduledExecutorService> decorator) Set up an additional ScheduledExecutorService decorator for a given key, even if that key is already present.
static void	setFactory(Schedulers.Factory factoryInstance) Replace Schedulers factories (newParallel, newSingle and newElastic).
static void	shutdownNow() Clear any cached Scheduler and call dispose on them.
static Scheduler	single() Scheduler that hosts a single-threaded ExecutorService-based worker and is suited for parallel work.
static Scheduler	single(Scheduler original) Wraps a single Scheduler.Worker from some other Scheduler and provides Scheduler.Worker services on top of it.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

DEFAULT_POOL_SIZE

public static final int DEFAULT_POOL_SIZE

Default pool size, initialized by system property `reactor.schedulers.defaultPoolSize` and falls back to the number of processors available to the runtime on init.

See Also:

Runtime.availableProcessors()

Constructor Detail

Schedulers

public Schedulers()

Method Detail

fromExecutor

public static Scheduler fromExecutor(Executor executor)

Create a Scheduler which uses a backing Executor to schedule Runnables for async operators.

Tasks scheduled with workers of this Scheduler are not guaranteed to run in FIFO order and strictly non-concurrently. If FIFO order is desired, use trampoline parameter of fromExecutor(Executor, boolean)

Parameters:

executor - an Executor

Returns:

a new Scheduler

fromExecutor

public static Scheduler fromExecutor(Executor executor,
                                     boolean trampoline)

Create a Scheduler which uses a backing Executor to schedule Runnables for async operators. Trampolining here means tasks submitted in a burst are queued by the Worker itself, which acts as a sole task from the perspective of the ExecutorService, so no reordering (but also no threading).

Parameters:

executor - an Executor

trampoline - set to false if this Scheduler is used by "operators" that already conflate Runnable executions (publishOn, subscribeOn...)

Returns:

a new Scheduler

fromExecutorService

public static Scheduler fromExecutorService(ExecutorService executorService)

Create a Scheduler which uses a backing ExecutorService to schedule Runnables for async operators.

Prefer using fromExecutorService(ExecutorService, String), especially if you plan on using metrics as this gives the executor a meaningful identifier.

Parameters:

executorService - an ExecutorService

Returns:

a new Scheduler

fromExecutorService

public static Scheduler fromExecutorService(ExecutorService executorService,
                                            String executorName)

Create a Scheduler which uses a backing ExecutorService to schedule Runnables for async operators.

Parameters:

executorService - an ExecutorService

Returns:

a new Scheduler

elastic

public static Scheduler elastic()

Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.

The maximum number of created thread pools is unbounded.

The default time-to-live for unused thread pools is 60 seconds, use the appropriate factory to set a different value.

This scheduler is not restartable.

Returns:

default instance of a Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work

parallel

public static Scheduler parallel()

Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work.

Returns:

default instance of a Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers

immediate

public static Scheduler immediate()

Executes tasks immediately instead of scheduling them.

As a consequence tasks run on the thread that submitted them (eg. the thread on which an operator is currently processing its onNext/onComplete/onError signals). This Scheduler is typically used as a "null object" for APIs that require a Scheduler but one doesn't want to change threads.

Returns:

a reusable Scheduler that executes tasks immediately instead of scheduling them

newElastic

public static Scheduler newElastic(String name)

Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.

The maximum number of created thread pools is unbounded.

The default time-to-live for unused thread pools is 60 seconds, use the appropriate factory to set a different value.

This scheduler is not restartable.

Parameters:

name - Thread prefix

Returns:

a new Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work

newElastic

public static Scheduler newElastic(String name,
                                   int ttlSeconds)

Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.

The maximum number of created thread pools is unbounded.

This scheduler is not restartable.

Parameters:

name - Thread prefix

ttlSeconds - Time-to-live for an idle Scheduler.Worker

Returns:

a new Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work

newElastic

public static Scheduler newElastic(String name,
                                   int ttlSeconds,
                                   boolean daemon)

Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.

The maximum number of created thread pools is unbounded.

This scheduler is not restartable.

Parameters:

name - Thread prefix

ttlSeconds - Time-to-live for an idle Scheduler.Worker

daemon - false if the Scheduler requires an explicit Scheduler.dispose() to exit the VM.

Returns:

a new Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work

newElastic

public static Scheduler newElastic(int ttlSeconds,
                                   ThreadFactory threadFactory)

Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.

The maximum number of created thread pools is unbounded.

This scheduler is not restartable.

Parameters:

ttlSeconds - Time-to-live for an idle Scheduler.Worker

threadFactory - a ThreadFactory to use each thread initialization

Returns:

a new Scheduler that dynamically creates ExecutorService-based Workers and caches the thread pools, reusing them once the Workers have been shut down.

newParallel

public static Scheduler newParallel(String name)

Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work. This type of Scheduler detects and rejects usage of blocking Reactor APIs.

Parameters:

name - Thread prefix

Returns:

a new Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work

newParallel

public static Scheduler newParallel(String name,
                                    int parallelism)

Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work. This type of Scheduler detects and rejects usage of blocking Reactor APIs.

Parameters:

name - Thread prefix

parallelism - Number of pooled workers.

Returns:

a new Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work

newParallel

public static Scheduler newParallel(String name,
                                    int parallelism,
                                    boolean daemon)

Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work. This type of Scheduler detects and rejects usage of blocking Reactor APIs.

Parameters:

name - Thread prefix

parallelism - Number of pooled workers.

daemon - false if the Scheduler requires an explicit Scheduler.dispose() to exit the VM.

Returns:

a new Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work

newParallel

public static Scheduler newParallel(int parallelism,
                                    ThreadFactory threadFactory)

Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work.

Parameters:

parallelism - Number of pooled workers.

threadFactory - a ThreadFactory to use for the fixed initialized number of Thread

Returns:

a new Scheduler that hosts a fixed pool of single-threaded ExecutorService-based workers and is suited for parallel work

newSingle

public static Scheduler newSingle(String name)

Scheduler that hosts a single-threaded ExecutorService-based worker and is suited for parallel work. This type of Scheduler detects and rejects usage * of blocking Reactor APIs.

Parameters:

name - Component and thread name prefix

Returns:

a new Scheduler that hosts a single-threaded ExecutorService-based worker

newSingle

public static Scheduler newSingle(String name,
                                  boolean daemon)

Scheduler that hosts a single-threaded ExecutorService-based worker and is suited for parallel work. This type of Scheduler detects and rejects usage of blocking Reactor APIs.

Parameters:

name - Component and thread name prefix

daemon - false if the Scheduler requires an explicit Scheduler.dispose() to exit the VM.

Returns:

a new Scheduler that hosts a single-threaded ExecutorService-based worker

newSingle

public static Scheduler newSingle(ThreadFactory threadFactory)

Scheduler that hosts a single-threaded ExecutorService-based worker and is suited for parallel work.

Parameters:

threadFactory - a ThreadFactory to use for the unique thread of the Scheduler

Returns:

a new Scheduler that hosts a single-threaded ExecutorService-based worker

onHandleError

public static void onHandleError(BiConsumer<Thread,? super Throwable> c)

Define a hook that is executed when a Scheduler has handled an error. Note that it is executed after the error has been passed to the thread uncaughtErrorHandler, which is not the case when a fatal error occurs (see Exceptions.throwIfJvmFatal(Throwable)).

Parameters:

c - the new hook to set.

isInNonBlockingThread

public static boolean isInNonBlockingThread()

Check if calling a Reactor blocking API in the current Thread is forbidden or not, by checking if the thread implements NonBlocking (in which case it is forbidden and this method returns true).

Returns:

true if blocking is forbidden in this thread, false otherwise

isNonBlockingThread

public static boolean isNonBlockingThread(Thread t)

Check if calling a Reactor blocking API in the given Thread is forbidden or not, by checking if the thread implements NonBlocking (in which case it is forbidden and this method returns true).

Returns:

true if blocking is forbidden in that thread, false otherwise

enableMetrics

public static void enableMetrics()

If Micrometer is available, set-up a decorator that will instrument any ExecutorService that backs a Scheduler. No-op if Micrometer isn't available. This instrumentation sends data to the Micrometer Global Registry.

Implementation Note:

Note that this is added as a decorator via Schedulers when enabling metrics for schedulers, which doesn't change the Factory.

disableMetrics

public static void disableMetrics()

If enableMetrics() has been previously called, removes the decorator. No-op if enableMetrics() hasn't been called.

resetFactory

public static void resetFactory()

Re-apply default factory to Schedulers

resetOnHandleError

public static void resetOnHandleError()

Reset the onHandleError(BiConsumer) hook to the default no-op behavior.

setFactory

public static void setFactory(Schedulers.Factory factoryInstance)

Replace Schedulers factories (newParallel, newSingle and newElastic). Also shutdown Schedulers from the cached factories (like single()) in order to also use these replacements, re-creating the shared schedulers from the new factory upon next use.

This method should be called safely and with caution, typically on app startup.

Parameters:

factoryInstance - an arbitrary Schedulers.Factory instance.

addExecutorServiceDecorator

public static boolean addExecutorServiceDecorator(String key,
                                                  BiFunction<Scheduler,ScheduledExecutorService,ScheduledExecutorService> decorator)

Set up an additional ScheduledExecutorService decorator for a given key only if that key is not already present. Note that the Schedulers.Factory's legacy Schedulers.Factory.decorateExecutorService(String, Supplier) method will always be invoked last after applying the decorators added via this method.

The decorator is a BiFunction taking the Scheduler and the backing ScheduledExecutorService as second argument. It returns the decorated ScheduledExecutorService.

Parameters:

key - the key under which to set up the decorator

decorator - the executor service decorator to add, if key not already present.

Returns:

true if the decorator was added, false if a decorator was already present for this key.

See Also:

setExecutorServiceDecorator(String, BiFunction), removeExecutorServiceDecorator(String)

setExecutorServiceDecorator

public static void setExecutorServiceDecorator(String key,
                                               BiFunction<Scheduler,ScheduledExecutorService,ScheduledExecutorService> decorator)

Set up an additional ScheduledExecutorService decorator for a given key, even if that key is already present. Note that the Schedulers.Factory's legacy Schedulers.Factory.decorateExecutorService(String, Supplier) method will always be invoked last after applying the decorators added via this method.

The decorator is a BiFunction taking the Scheduler and the backing ScheduledExecutorService as second argument. It returns the decorated ScheduledExecutorService.

Parameters:

key - the key under which to set up the decorator

decorator - the executor service decorator to add, if key not already present.

See Also:

addExecutorServiceDecorator(String, BiFunction), removeExecutorServiceDecorator(String)

removeExecutorServiceDecorator

public static BiFunction<Scheduler,ScheduledExecutorService,ScheduledExecutorService> removeExecutorServiceDecorator(String key)

Remove an existing ScheduledExecutorService decorator if it has been set up via addExecutorServiceDecorator(String, BiFunction). Note that the Schedulers.Factory's legacy Schedulers.Factory.decorateExecutorService(String, Supplier) method is always applied last, even if all other decorators have been removed via this method.

In case the decorator implements Disposable, it is also disposed.

Parameters:

key - the key for the executor service decorator to remove

Returns:

the removed decorator, or null if none was set for that key

See Also:

addExecutorServiceDecorator(String, BiFunction), setExecutorServiceDecorator(String, BiFunction)

decorateExecutorService

public static ScheduledExecutorService decorateExecutorService(Scheduler owner,
                                                               ScheduledExecutorService original)

This method is aimed at Scheduler implementors, enabling custom implementations that are backed by a ScheduledExecutorService to also have said executors decorated (ie. for instrumentation purposes).

It applies the decorators added via addExecutorServiceDecorator(String, BiFunction), so it shouldn't be added as a decorator. Note also that decorators are not guaranteed to be idempotent, so this method should be called only once per executor.

Parameters:

owner - a Scheduler that owns the ScheduledExecutorService

original - the ScheduledExecutorService that the Scheduler wants to use originally

Returns:

the decorated ScheduledExecutorService, or the original if no decorator is set up

See Also:

addExecutorServiceDecorator(String, BiFunction), removeExecutorServiceDecorator(String)

shutdownNow

public static void shutdownNow()

Clear any cached Scheduler and call dispose on them.

single

public static Scheduler single()

Scheduler that hosts a single-threaded ExecutorService-based worker and is suited for parallel work. Will cache the returned schedulers for subsequent calls until dispose.

Returns:

default instance of a Scheduler that hosts a single-threaded ExecutorService-based worker

single

public static Scheduler single(Scheduler original)

Wraps a single Scheduler.Worker from some other Scheduler and provides Scheduler.Worker services on top of it.

Use the Scheduler.dispose() to release the wrapped worker.

Parameters:

original - a Scheduler to call upon to get the single Scheduler.Worker

Returns:

a wrapping Scheduler consistently returning a same worker from a source Scheduler