HazelcastAPI Ringbuffer Arayüzü

Giriş

Şu satırı dahil ederiz

import com.hazelcast.ringbuffer.Ringbuffer;

Açıklaması şöyle. Yani tüm veri tek bir node ve replica node üzerindedir.

Hazelcast Ringbuffer is a replicated but not partitioned data structure that stores its data in a ring-like structure. You can think of it as a circular array with a given capacity. 

readManyAsync metodu

İmzası şöyle

CompletionStage<ReadResultSet<E>> readManyAsync(
  long startSequence,
  int minCount,
  int maxCount,
  IFunction<E, Boolean> filter);

Açıklaması şöyle

startSequence: Sequence of the first item to read.

minCount: Minimum number of items to read. If you do not want to block, set it to 0. If you want to block for at least one item, set it to 1.

maxCount: Maximum number of the items to retrieve. Its value cannot exceed 1000.

filter: A function that accepts an item and checks if it should be returned. If no filtering should be applied, set it to null.

Örnek

Şöyle yaparız. Burada sequence farklı artırılıyor çünkü, filtre ile bazı şeyler süzülmüş olabilir

long sequence = rb.headSequence();
for(;;) {
  CompletionStage<ReadResultSet<String>> f = rb.readManyAsync(sequence, 1, 10, null);
  CompletionStage<Integer> readCountStage = f.thenApplyAsync(rs -> {
    for (String s : rs) {
      System.out.println(s);
    }
    return rs.readCount();
  });
  sequence += readCountStage.toCompletableFuture().join();
}

readOne metodu

Açıklaması şöyle

Use the method readOne to return the item at the given sequence; readOne blocks if no item is available. To read the next item, increment the sequence by one.

By exposing the sequence, you can now move the item from the Ringbuffer as long as the item is still available. If the item is not available any longer, StaleSequenceException is thrown.

Örnek

Şöyle yaparız

HazelcastInstance hz = Hazelcast.newHazelcastInstance();
Ringbuffer<String> ringbuffer = hz.getRingbuffer("rb");
long sequence = ringbuffer.headSequence();
while(true){
  String item = ringbuffer.readOne(sequence);
  sequence++;
  // process item
}