E
- Type of dispatched signalpublic final class RingBufferWorkProcessor<E> extends ExecutorPoweredProcessor<E,E>
The processor is very similar to RingBufferProcessor
but only partially respects the
Reactive Streams contract.
The purpose of this processor is to distribute the signals to only one of the subscribed subscribers and to share the demand amongst all subscribers. The scenario is akin to Executor or Round-Robin distribution. However there is no guarantee the distribution will be respecting a round-robin distribution all the time.
The core use for this component is to scale up easily without suffering the overhead of an Executor and without using dedicated queues by subscriber, which is less used memory, less GC, more win.
executor
autoCancel, SMALL_BUFFER_SIZE, SUBSCRIBER_COUNT, upstreamSubscription
Modifier and Type | Method and Description |
---|---|
static <E> RingBufferWorkProcessor<E> |
create()
Create a new RingBufferWorkProcessor using
ReactorProcessor.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy
and auto-cancel. |
static <E> RingBufferWorkProcessor<E> |
create(boolean autoCancel)
Create a new RingBufferWorkProcessor using
ReactorProcessor.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy
and the passed auto-cancel setting. |
static <E> RingBufferWorkProcessor<E> |
create(ExecutorService service)
Create a new RingBufferWorkProcessor using
ReactorProcessor.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy
and auto-cancel. |
static <E> RingBufferWorkProcessor<E> |
create(ExecutorService service,
boolean autoCancel)
Create a new RingBufferWorkProcessor using
ReactorProcessor.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy
and the passed auto-cancel setting. |
static <E> RingBufferWorkProcessor<E> |
create(ExecutorService service,
int bufferSize)
Create a new RingBufferProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
create(ExecutorService service,
int bufferSize,
boolean autoCancel)
Create a new RingBufferWorkProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
create(ExecutorService executor,
int bufferSize,
com.lmax.disruptor.WaitStrategy strategy)
Create a new RingBufferWorkProcessor using the passed buffer size and blockingWait Strategy settings
but will auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
create(ExecutorService executor,
int bufferSize,
com.lmax.disruptor.WaitStrategy strategy,
boolean autoCancel)
Create a new RingBufferWorkProcessor using the passed buffer size, wait strategy
and auto-cancel settings.
|
static <E> RingBufferWorkProcessor<E> |
create(String name,
int bufferSize)
Create a new RingBufferProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
create(String name,
int bufferSize,
boolean autoCancel)
Create a new RingBufferProcessor using the passed buffer size, blockingWait Strategy
and the passed auto-cancel setting.
|
static <E> RingBufferWorkProcessor<E> |
create(String name,
int bufferSize,
com.lmax.disruptor.WaitStrategy strategy)
Create a new RingBufferWorkProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
create(String name,
int bufferSize,
com.lmax.disruptor.WaitStrategy strategy,
boolean autoCancel)
Create a new RingBufferWorkProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel settings.
|
long |
getAvailableCapacity() |
long |
getCapacity()
Return defined element capacity, used to drive new
Subscription
request needs. |
void |
onComplete() |
void |
onError(Throwable t) |
void |
onNext(E o) |
static <E> RingBufferWorkProcessor<E> |
share()
Create a new RingBufferWorkProcessor using
ReactorProcessor.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy
and auto-cancel. |
static <E> RingBufferWorkProcessor<E> |
share(boolean autoCancel)
Create a new RingBufferWorkProcessor using
ReactorProcessor.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy
and the passed auto-cancel setting. |
static <E> RingBufferWorkProcessor<E> |
share(ExecutorService service)
Create a new RingBufferWorkProcessor using
ReactorProcessor.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy
and auto-cancel. |
static <E> RingBufferWorkProcessor<E> |
share(ExecutorService service,
boolean autoCancel)
Create a new RingBufferWorkProcessor using
ReactorProcessor.SMALL_BUFFER_SIZE backlog size, blockingWait Strategy
and the passed auto-cancel setting. |
static <E> RingBufferWorkProcessor<E> |
share(ExecutorService service,
int bufferSize)
Create a new RingBufferProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
share(ExecutorService service,
int bufferSize,
boolean autoCancel)
Create a new RingBufferWorkProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
share(ExecutorService executor,
int bufferSize,
com.lmax.disruptor.WaitStrategy strategy)
Create a new RingBufferWorkProcessor using the passed buffer size and blockingWait Strategy settings
but will auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
share(ExecutorService executor,
int bufferSize,
com.lmax.disruptor.WaitStrategy strategy,
boolean autoCancel)
Create a new RingBufferWorkProcessor using the passed buffer size, wait strategy
and auto-cancel settings.
|
static <E> RingBufferWorkProcessor<E> |
share(String name,
int bufferSize)
Create a new RingBufferProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
share(String name,
int bufferSize,
boolean autoCancel)
Create a new RingBufferProcessor using the passed buffer size, blockingWait Strategy
and the passed auto-cancel setting.
|
static <E> RingBufferWorkProcessor<E> |
share(String name,
int bufferSize,
com.lmax.disruptor.WaitStrategy strategy)
Create a new RingBufferWorkProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel.
|
static <E> RingBufferWorkProcessor<E> |
share(String name,
int bufferSize,
com.lmax.disruptor.WaitStrategy strategy,
boolean autoCancel)
Create a new RingBufferWorkProcessor using the passed buffer size, blockingWait Strategy
and auto-cancel settings.
|
void |
subscribe(org.reactivestreams.Subscriber<? super E> subscriber) |
String |
toString() |
org.reactivestreams.Publisher<Void> |
writeWith(org.reactivestreams.Publisher<? extends E> source) |
alive, awaitAndShutdown, awaitAndShutdown, forceShutdown, shutdown
accept, decrementSubscribers, incrementSubscribers, isReactivePull, onSubscribe
public static <E> RingBufferWorkProcessor<E> create()
ReactorProcessor.SMALL_BUFFER_SIZE
backlog size, blockingWait Strategy
and auto-cancel.
A new Cached ThreadExecutorPool will be implicitely created.
E
- Type of processed signalspublic static <E> RingBufferWorkProcessor<E> create(boolean autoCancel)
ReactorProcessor.SMALL_BUFFER_SIZE
backlog size, blockingWait Strategy
and the passed auto-cancel setting.
A new Cached ThreadExecutorPool will be implicitely created.
E
- Type of processed signalsautoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> create(ExecutorService service)
ReactorProcessor.SMALL_BUFFER_SIZE
backlog size, blockingWait Strategy
and auto-cancel.
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.E
- Type of processed signalsservice
- A provided ExecutorService to manage threading infrastructurepublic static <E> RingBufferWorkProcessor<E> create(ExecutorService service, boolean autoCancel)
ReactorProcessor.SMALL_BUFFER_SIZE
backlog size, blockingWait Strategy
and the passed auto-cancel setting.
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsservice
- A provided ExecutorService to manage threading infrastructureautoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> create(String name, int bufferSize)
A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.
E
- Type of processed signalsname
- Use a new Cached ExecutorService and assign this name to the created threadsbufferSize
- A Backlog Size to mitigate slow subscriberspublic static <E> RingBufferWorkProcessor<E> create(String name, int bufferSize, boolean autoCancel)
A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.
E
- Type of processed signalsname
- Use a new Cached ExecutorService and assign this name to the created threadsbufferSize
- A Backlog Size to mitigate slow subscribersautoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> create(ExecutorService service, int bufferSize)
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsservice
- A provided ExecutorService to manage threading infrastructurebufferSize
- A Backlog Size to mitigate slow subscriberspublic static <E> RingBufferWorkProcessor<E> create(ExecutorService service, int bufferSize, boolean autoCancel)
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsservice
- A provided ExecutorService to manage threading infrastructurebufferSize
- A Backlog Size to mitigate slow subscribersautoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> create(String name, int bufferSize, com.lmax.disruptor.WaitStrategy strategy)
A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.
E
- Type of processed signalsname
- Use a new Cached ExecutorService and assign this name to the created threadsbufferSize
- A Backlog Size to mitigate slow subscribersstrategy
- A RingBuffer WaitStrategy to use instead of the default LiteBlockingWaitStrategy.public static <E> RingBufferWorkProcessor<E> create(String name, int bufferSize, com.lmax.disruptor.WaitStrategy strategy, boolean autoCancel)
A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.
E
- Type of processed signalsname
- Use a new Cached ExecutorService and assign this name to the created threadsbufferSize
- A Backlog Size to mitigate slow subscribersstrategy
- A RingBuffer WaitStrategy to use instead of the default LiteBlockingWaitStrategy.autoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> create(ExecutorService executor, int bufferSize, com.lmax.disruptor.WaitStrategy strategy)
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsexecutor
- A provided ExecutorService to manage threading infrastructurebufferSize
- A Backlog Size to mitigate slow subscribersstrategy
- A RingBuffer WaitStrategy to use instead of the default LiteBlockingWaitStrategy.public static <E> RingBufferWorkProcessor<E> create(ExecutorService executor, int bufferSize, com.lmax.disruptor.WaitStrategy strategy, boolean autoCancel)
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsexecutor
- A provided ExecutorService to manage threading infrastructurebufferSize
- A Backlog Size to mitigate slow subscribersstrategy
- A RingBuffer WaitStrategy to use instead of the default LiteBlockingWaitStrategy.autoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> share()
ReactorProcessor.SMALL_BUFFER_SIZE
backlog size, blockingWait Strategy
and auto-cancel.
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
A new Cached ThreadExecutorPool will be implicitely created.
E
- Type of processed signalspublic static <E> RingBufferWorkProcessor<E> share(boolean autoCancel)
ReactorProcessor.SMALL_BUFFER_SIZE
backlog size, blockingWait Strategy
and the passed auto-cancel setting.
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
A new Cached ThreadExecutorPool will be implicitely created.
E
- Type of processed signalsautoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> share(ExecutorService service)
ReactorProcessor.SMALL_BUFFER_SIZE
backlog size, blockingWait Strategy
and auto-cancel.
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.E
- Type of processed signalsservice
- A provided ExecutorService to manage threading infrastructurepublic static <E> RingBufferWorkProcessor<E> share(ExecutorService service, boolean autoCancel)
ReactorProcessor.SMALL_BUFFER_SIZE
backlog size, blockingWait Strategy
and the passed auto-cancel setting.
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsservice
- A provided ExecutorService to manage threading infrastructureautoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> share(String name, int bufferSize)
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.
E
- Type of processed signalsname
- Use a new Cached ExecutorService and assign this name to the created threadsbufferSize
- A Backlog Size to mitigate slow subscriberspublic static <E> RingBufferWorkProcessor<E> share(String name, int bufferSize, boolean autoCancel)
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.
E
- Type of processed signalsname
- Use a new Cached ExecutorService and assign this name to the created threadsbufferSize
- A Backlog Size to mitigate slow subscribersautoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> share(ExecutorService service, int bufferSize)
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsservice
- A provided ExecutorService to manage threading infrastructurebufferSize
- A Backlog Size to mitigate slow subscriberspublic static <E> RingBufferWorkProcessor<E> share(ExecutorService service, int bufferSize, boolean autoCancel)
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsservice
- A provided ExecutorService to manage threading infrastructurebufferSize
- A Backlog Size to mitigate slow subscribersautoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> share(String name, int bufferSize, com.lmax.disruptor.WaitStrategy strategy)
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.
E
- Type of processed signalsname
- Use a new Cached ExecutorService and assign this name to the created threadsbufferSize
- A Backlog Size to mitigate slow subscribersstrategy
- A RingBuffer WaitStrategy to use instead of the default LiteBlockingWaitStrategy.public static <E> RingBufferWorkProcessor<E> share(String name, int bufferSize, com.lmax.disruptor.WaitStrategy strategy, boolean autoCancel)
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
A new Cached ThreadExecutorPool will be implicitely created and will use the passed name to qualify the created threads.
E
- Type of processed signalsname
- Use a new Cached ExecutorService and assign this name to the created threadsbufferSize
- A Backlog Size to mitigate slow subscribersstrategy
- A RingBuffer WaitStrategy to use instead of the default LiteBlockingWaitStrategy.autoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public static <E> RingBufferWorkProcessor<E> share(ExecutorService executor, int bufferSize, com.lmax.disruptor.WaitStrategy strategy)
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsexecutor
- A provided ExecutorService to manage threading infrastructurebufferSize
- A Backlog Size to mitigate slow subscribersstrategy
- A RingBuffer WaitStrategy to use instead of the default LiteBlockingWaitStrategy.public static <E> RingBufferWorkProcessor<E> share(ExecutorService executor, int bufferSize, com.lmax.disruptor.WaitStrategy strategy, boolean autoCancel)
A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that will fan-in data.
The passed ExecutorService
will execute as many event-loop
consuming the ringbuffer as subscribers.
E
- Type of processed signalsexecutor
- A provided ExecutorService to manage threading infrastructurebufferSize
- A Backlog Size to mitigate slow subscribersstrategy
- A RingBuffer WaitStrategy to use instead of the default LiteBlockingWaitStrategy.autoCancel
- Should this propagate cancellation when unregistered by all subscribers ?public void subscribe(org.reactivestreams.Subscriber<? super E> subscriber)
public void onNext(E o)
public void onError(Throwable t)
public void onComplete()
onComplete
in interface org.reactivestreams.Subscriber<E>
onComplete
in class ExecutorPoweredProcessor<E,E>
public org.reactivestreams.Publisher<Void> writeWith(org.reactivestreams.Publisher<? extends E> source)
public long getCapacity()
NonBlocking
Subscription
request needs. This is the maximum in-flight data allowed to transit to this elements.getCapacity
in interface NonBlocking
getCapacity
in class ReactorProcessor<E,E>
public long getAvailableCapacity()
getAvailableCapacity
in class ReactorProcessor<E,E>
Copyright © 2017. All rights reserved.