import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
public class LowGCMemoryQueueExample {
public static class Event {
private String data;
public void setData(String data) {
this.data = data;
}
public String getData() {
return data;
}
}
public static class EventFactory implements EventFactory<Event> {
@Override
public Event newInstance() {
return new Event();
}
}
public static class EventHandler implements EventHandler<Event> {
@Override
public void onEvent(Event event, long sequence, boolean endOfBatch) throws Exception {
System.out.println("Processing event: " + event.getData());
}
}
public static void main(String[] args) {
RingBuffer<Event> ringBuffer = RingBuffer.create(ProducerType.SINGLE, new EventFactory(), 1024, new YieldingWaitStrategy());
ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
EventHandler eventHandler = new EventHandler();
BatchEventProcessor<Event> eventProcessor = new BatchEventProcessor<>(ringBuffer, sequenceBarrier, eventHandler);
executor.submit(eventProcessor);
for (int i = 0; i < 10; i++) {
long sequence = ringBuffer.next();
Event event = ringBuffer.get(sequence);
event.setData("Event " + i);
ringBuffer.publish(sequence);
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
eventProcessor.halt();
executor.shutdown();
}
}
