E
- The type of the elements that the Ringbuffer containspublic interface Ringbuffer<E> extends DistributedObject
The ringbuffer has 2 always incrementing sequences:
tailSequence()
: this is the side where the youngest item is found.
So the tail is the side of the ringbuffer where items are added to.
headSequence()
: this is the side where the oldest items are found.
So the head is the side where items get discarded.
If data is read from a ringbuffer with a sequence that is smaller than the
headSequence, it means that the data is not available anymore and a
StaleSequenceException
is thrown.
A Ringbuffer currently is a replicated, but not partitioned data structure.
So all data is stored in a single partition, similarly to the IQueue
implementation.
A Ringbuffer can be used in a way similar to the IQueue, but one of the key
differences is that a queue.take
is destructive, meaning that only 1
thread is able to take an item. A ringbuffer.read
is not
destructive, so you can have multiple threads reading the same item multiple
times.
The Ringbuffer is the backing data structure for the reliable
ITopic
implementation. See
ReliableTopicConfig
.
A Ringbuffer can be configured to be backed by a
RingbufferStore
. All write methods will delegate
to the store to persist the items, while reader methods will try to read
items from the store if not found in the in-memory Ringbuffer.
When a Ringbuffer is constructed with a backing store, head and tail sequences are set to the following
tailSequence
: lastStoreSequence
headSequence
: lastStoreSequence
+ 1lastStoreSequence
is the sequence of the previously last
stored item.
Supports split brain protection SplitBrainProtectionConfig
since 3.10 in
cluster versions 3.10 and higher.
Asynchronous methods
Asynchronous methods return a CompletionStage
that can be used to
chain further computation stages. Alternatively, a CompletableFuture
can be obtained via CompletionStage.toCompletableFuture()
to wait
for the operation to complete in a blocking way.
Actions supplied for dependent completions of default non-async methods and async methods
without an explicit Executor
argument are performed
by the ForkJoinPool.commonPool()
(unless it does not
support a parallelism level of at least 2, in which case a new Thread
is
created per task).
Modifier and Type | Method and Description |
---|---|
long |
add(E item)
Adds an item to the tail of the Ringbuffer.
|
CompletionStage<Long> |
addAllAsync(Collection<? extends E> collection,
OverflowPolicy overflowPolicy)
Adds all the items of a collection to the tail of the Ringbuffer.
|
CompletionStage<Long> |
addAsync(E item,
OverflowPolicy overflowPolicy)
Asynchronously writes an item with a configurable
OverflowPolicy . |
long |
capacity()
Returns the capacity of this Ringbuffer.
|
long |
headSequence()
Returns the sequence of the head.
|
CompletionStage<ReadResultSet<E>> |
readManyAsync(long startSequence,
int minCount,
int maxCount,
IFunction<E,Boolean> filter)
Reads a batch of items from the Ringbuffer.
|
E |
readOne(long sequence)
Reads one item from the Ringbuffer.
|
long |
remainingCapacity()
Returns the remaining capacity of the ringbuffer.
|
long |
size()
Returns number of items in the Ringbuffer.
|
long |
tailSequence()
Returns the sequence of the tail.
|
destroy, getDestroyContextForTenant, getName, getPartitionKey, getServiceName
long capacity()
long size()
If no TTL is set, the size will always be equal to capacity after the head completed the first loop around the ring. This is because no items are being removed.
long tailSequence()
The initial value of the tail is -1 if the Ringbuffer is not backed by a store, otherwise tail sequence will be set to the sequence of the previously last stored item.
long headSequence()
If the RingBuffer is empty, the head will be one more than the tail.
The initial value of the head is 0 if the Ringbuffer is not backed by a store, otherwise head sequence will be set to the sequence of the previously last stored item + 1. In both cases head sequence is 1 more than the tail sequence.
long remainingCapacity()
If TTL is disabled, the remaining capacity is equal to the total ringbuffer capacity.
RingbufferConfig.DEFAULT_TTL_SECONDS
,
capacity()
long add(@Nonnull E item)
addAsync(Object, OverflowPolicy)
and the OverflowPolicy
.
The returned value is the sequence of the added item. Using this sequence you can read the added item.
If the Ringbuffer is backed by a RingbufferStore
,
the item gets persisted by the underlying store via
RingbufferStore.store(long, Object)
. Note that
in case an exception is thrown by the store, it prevents the item from being
added to the Ringbuffer, keeping the store, primary and the backups
consistent.
item
- the item to add.NullPointerException
- if item is null.addAsync(Object, OverflowPolicy)
CompletionStage<Long> addAsync(@Nonnull E item, @Nonnull OverflowPolicy overflowPolicy)
OverflowPolicy
.
If there is space in the Ringbuffer, the call will return the sequence of the written item. If there is no space, it depends on the overflow policy what happens:
OverflowPolicy.OVERWRITE
: we just overwrite the oldest item
in the Ringbuffer and we violate the TTLOverflowPolicy.FAIL
: we return -1 The reason that FAIL exist is to give the opportunity to obey the TTL. If blocking behavior is required, this can be implemented using retrying in combination with an exponential backoff. Example:
long sleepMs = 100;
for (; ; ) {
long result = ringbuffer.addAsync(item, FAIL).toCompletableFuture().get();
if (result != -1) {
break;
}
TimeUnit.MILLISECONDS.sleep(sleepMs);
sleepMs = min(5000, sleepMs * 2);
}
If the Ringbuffer is backed by a RingbufferStore
,
the item gets persisted by the underlying store via
RingbufferStore.store(long, Object)
. Note
that in case an exception is thrown by the store, it prevents the item
from being added to the Ringbuffer, keeping the store, primary and the
backups consistent.
item
- the item to addoverflowPolicy
- the OverflowPolicy to use.NullPointerException
- if item or overflowPolicy is null.E readOne(long sequence) throws InterruptedException
If the sequence is one beyond the current tail, this call blocks until an item is added. This means that the ringbuffer can be processed using the following idiom:
Ringbuffer<String> ringbuffer = hz.getRingbuffer("rb");
long seq = ringbuffer.headSequence();
while(true){
String item = ringbuffer.readOne(seq);
seq++;
... process item
}
This method is not destructive unlike e.g. a BaseQueue.take()
.
So the same item can be read by multiple readers or it can be read
multiple times by the same reader.
Currently it isn't possible to control how long this call is going to
block. In the future we could add e.g.
tryReadOne(long sequence, long timeout, TimeUnit unit)
.
If the item is not in the Ringbuffer an attempt is made to read it from
the underlying RingbufferStore
via
RingbufferStore.load(long)
if store is
configured for the Ringbuffer. These cases may increase the execution time
significantly depending on the implementation of the store. Note that
exceptions thrown by the store are propagated to the caller.
sequence
- the sequence of the item to read.StaleSequenceException
- if the sequence is smaller than
headSequence()
. Because a
Ringbuffer won't store all event
indefinitely, it can be that the data
for the given sequence doesn't exist
anymore and the
StaleSequenceException
is thrown.
It is up to the caller to deal with
this particular situation, e.g. throw an
Exception or restart from the last known
head. That is why the
StaleSequenceException contains the last
known head.IllegalArgumentException
- if sequence is smaller than 0 or larger than tailSequence()
+1.InterruptedException
- if the call is interrupted while blocking.CompletionStage<Long> addAllAsync(@Nonnull Collection<? extends E> collection, @Nonnull OverflowPolicy overflowPolicy)
An addAll is likely to outperform multiple calls to add(Object)
due to better io utilization and a reduced number of executed operations.
If the batch is empty, the call is ignored.
When the collection is not empty, the content is copied into a different data-structure. This means that:
If the collection is larger than the capacity of the Ringbuffer, then the items that were written first will be overwritten. Therefore this call will not block.
The items are inserted in the order of the Iterator of the collection. If an addAll is executed concurrently with an add or addAll, no guarantee is given that items are contiguous.
The result of the future contains the sequenceId of the last written item.
If the Ringbuffer is backed by a RingbufferStore
,
the items are persisted by the underlying store via
RingbufferStore.storeAll(long, Object[])
.
Note that in case an exception is thrown by the store, it makes the
Ringbuffer not adding any of the items to the primary and the backups.
Keeping the store consistent with the primary and the backups is the
responsibility of the store.
collection
- the batch of items to add.NullPointerException
- if batch is null, or if an item in this
batch is null or if overflowPolicy is nullIllegalArgumentException
- if collection is emptyCompletionStage<ReadResultSet<E>> readManyAsync(long startSequence, int minCount, int maxCount, @Nullable IFunction<E,Boolean> filter)
maxCount
,
these items are returned. So it could be the number of items read is
smaller than the maxCount
.
If there are less items available than minCount
, then this call
blocks.
Warning:
These blocking calls consume server memory and if there are many calls, it can be possible to see leaking memory or OOME.
Reading a batch of items is likely to perform better because less overhead is involved.
A filter can be provided to only select items that need to be read. If the filter is null, all items are read. If the filter is not null, only items where the filter function returns true are returned. Using filters is a good way to prevent getting items that are of no value to the receiver. This reduces the amount of IO and the number of operations being executed, and can result in a significant performance improvement.
For each item not available in the Ringbuffer an attempt is made to read
it from the underlying RingbufferStore
via
multiple invocations of RingbufferStore.load(long)
,
if store is configured for the Ringbuffer. These cases may increase the
execution time significantly depending on the implementation of the store.
Note that exceptions thrown by the store are propagated to the caller.
If the startSequence is smaller than the smallest sequence still available
in the Ringbuffer (headSequence()
, then the smallest available
sequence will be used as the start sequence and the minimum/maximum
number of items will be attempted to be read from there on.
If the startSequence is bigger than the last available sequence in the
Ringbuffer (tailSequence()
), then the last available sequence
plus one will be used as the start sequence and the call will block
until further items become available and it can read at least the
minimum number of items.
startSequence
- the startSequence of the first item to read.minCount
- the minimum number of items to read.maxCount
- the maximum number of items to read.filter
- the filter. Filter is allowed to be null, indicating
there is no filter.IllegalArgumentException
- if startSequence is smaller than 0
or if minCount smaller than 0
or if minCount larger than maxCount,
or if maxCount larger than the capacity of the ringbuffer
or if maxCount larger than 1000 (to prevent overload)Copyright © 2022 Hazelcast, Inc.. All rights reserved.