Ringbuffer: новая структура данных в Hazelcast

Круто то, что возвращаемую последовательность можно использовать и для считывания элемента:

String item = rb.readOne(sequence);

Поскольку каждый элемент уникально идентифицируется по его идентификатору последовательности, возвращаемый идентификатор последовательности является уникальным и может использоваться как дешевый генератор идентификаторов, если вы используете Ringbuffer.

Кольцевой буфер по сравнению с очередью

Хорошая вещь в Ringbuffer, по сравнению с очередью, состоит в том, что с очередью получение является разрушительной операцией; таким образом, только один поток может взять определенный элемент из очереди. Как только это взято, это ушло. Это может быть проблематично по двум причинам:

1. Что происходит, когда происходит сбой системы после того, как предмет был взят, но до того, как он был полностью обработан?
2. Что произойдет, если вы хотите, чтобы несколько читателей прочитали один и тот же элемент? Один из подходов заключается в создании очереди для каждого читателя и внесении в каждую очередь. Проблема в том, что он делает путы очень дорогими, потому что с N читателями нужно делать N путов.

Поскольку чтение в кольцевом буфере не является разрушительной операцией, и читатель контролирует, какие элементы он хочет прочитать, читателю легко реализовать гарантии доставки, сохраняя идентификатор последовательности.

• По крайней мере, один раз : сохранить идентификатор последовательности после того, как элемент был полностью обработан. Если происходит сбой системы до того, как элемент был полностью обработан, тот же элемент будет прочитан снова, поскольку сохраненный идентификатор последовательности все еще содержит старое значение.
• Самое большее один раз : сохранить идентификатор последовательности до начала обработки элемента. Если происходит сбой системы до полной обработки элемента, загружается идентификатор последовательности элемента, который мы потенциально не смогли обработать, и система может продолжить со следующего элемента.

Другое большое преимущество операции чтения, которая не является разрушительной, заключается в том, что она очень быстрая, поскольку ее не нужно тиражировать — в отличие от очереди.

Вместимость

Каждый Ringbuffer создается с определенной емкостью — по умолчанию 10 тыс. Элементов. Ringbuffer не может вырасти выше этой емкости, поэтому самые старые предметы в конечном итоге перезаписываются (подробнее об этом ниже). Кольцевой буфер может быть настроен с использованием XML или нашего программного API. Если мы хотим установить емкость:

RingbufferConfig rbConfig = new RingbufferConfig("rb")
    .setCapacity(50 * 1000);
Config config = new Config();
config.addRingbufferConfig(rbConfig);
HazelcastInstance hz = Hazelcast.newHazelcastInstance(config);
Ringbuffer<String&gr; rb = hz.getRingbuffer("rb");

RingbufferConfig rbConfig = new RingbufferConfig("rb")
    .setCapacity(50 * 1000)
    .setTimeToLiveSeconds(30);

long sleepMs = 100;
for (; ; ) {
    long result = ringbuffer.addAsync(item, OverflowPolicy.FAIL).get();
    if (result != -1) {
        break;
    }

    TimeUnit.MILLISECONDS.sleep(sleepMs);
    sleepMs = min(5000, sleepMs * 2);
}

List<String> items = Arrays.asList("1","2","3");
ICompletableFuture<Long> f = rb.addAllAsync(items, OverflowPolicy.OVERWRITE);
f.get()

long sequence = rb.headSequence();
for(;;) {

    ICompletableFuture<ReadResultSet<String>> f = rb.readManyAsync(sequence, 1, 100, null);
    ReadResultSet<String> rs = f.get();
    for (String s : rs) {
        System.out.println(s);
    }
    sequence+=rs.readCount();
}

public class EngineerFilter<Employee, Boolean> implements Filter {
    Boolean apply(Employee e){
    return e instanceof Engineer;
    }
}