// declareBufferTrigger<String> buffer = SimpleBufferTrigger.<String, Set<String>> newBuilder() .triggerStrategy(newMultiIntervalTriggerStrategy() .on(10, SECONDS, 1) .on(5, SECONDS, 10) .on(1, SECONDS, 100) ) .consumer(this::out) .setContainer(ConcurrentSkipListSet::new, Set::add) // default is Collections.newSetFromMap(new ConcurrentHashMap<>()) .build(); // enqueuebuffer.enqueue("i'm ok"); // consumer declareprivatevoidout(Collection<String> set){set.forEach(System.out::println)} // batch consumer blocking queueBufferTrigger<String> buffer = BatchConsumeBlockingQueueTrigger.<String> newBuilder() .batchSize(3) .setConsumerEx(this::out) .bufferSize(5) .linger(ofSeconds(2)) .build();