Understanding and Implementing LMAX Disruptor in Java

Background

In a project where Disruptor is used as a message queue, the author records the experience, noting that Disruptor is fast, open‑source, and not a traditional broker like Kafka or RabbitMQ.

Disruptor Introduction

Disruptor is a high‑performance queue developed by LMAX to solve memory‑queue latency, achieving throughput comparable to I/O operations.

Based on Disruptor, a single‑threaded system can handle 6 million orders per second; it gained industry attention after a 2010 QCon talk.

It is an open‑source Java framework designed for the producer‑consumer problem, providing a bounded queue with very low latency and high throughput.

Disruptor is essentially a design pattern for concurrent, buffered, producer‑consumer scenarios, offering a significant performance boost.

GitHub repository: https://github.com/LMAX-Exchange/disruptor

Core Concepts

The following domain objects map directly to the code implementation:

Ring Buffer – the circular buffer that stores events.

Sequence – sequential numbers used to track event progress.

Sequencer – core component with SingleProducerSequencer and MultiProducerSequencer implementations.

Sequence Barrier – controls visibility of published sequences to consumers.

Wait Strategy – defines how consumers wait for the next event.

Event – the data exchanged between producer and consumer.

EventProcessor – holds consumer sequence and runs the event loop.

EventHandler – user‑implemented interface to process events.

Producer – code that publishes events to the Disruptor.

Demo – Step‑by‑Step Implementation

1. Add Maven dependency:

<dependency>
  <groupId>com.lmax</groupId>
  <artifactId>disruptor</artifactId>
  <version>3.4.4</version>
</dependency>

2. Define the message model:

@Data
public class MessageModel {
    private String message;
}

3. Create an EventFactory :

public class HelloEventFactory implements EventFactory
{
    @Override
    public MessageModel newInstance() {
        return new MessageModel();
    }
}

4. Implement the consumer EventHandler :

@Slf4j
public class HelloEventHandler implements EventHandler
{
    @Override
    public void onEvent(MessageModel event, long sequence, boolean endOfBatch) {
        try {
            Thread.sleep(1000);
            log.info("Consumer start processing");
            if (event != null) {
                log.info("Consumed message: {}", event);
            }
        } catch (Exception e) {
            log.info("Consumer processing failed");
        }
        log.info("Consumer processing finished");
    }
}

5. Provide a bean manager to access Spring beans:

@Component
public class BeanManager implements ApplicationContextAware {
    private static ApplicationContext applicationContext = null;
    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        this.applicationContext = ctx;
    }
    public static ApplicationContext getApplicationContext() { return applicationContext; }
    public static Object getBean(String name) { return applicationContext.getBean(name); }
    public static
T getBean(Class
clazz) { return applicationContext.getBean(clazz); }
}

6. Configure the Disruptor bean:

@Configuration
public class MQManager {
    @Bean("messageModel")
    public RingBuffer
messageModelRingBuffer() {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        HelloEventFactory factory = new HelloEventFactory();
        int bufferSize = 1024 * 256;
        Disruptor
disruptor = new Disruptor<>(factory, bufferSize, executor,
                ProducerType.SINGLE, new BlockingWaitStrategy());
        disruptor.handleEventsWith(new HelloEventHandler());
        disruptor.start();
        return disruptor.getRingBuffer();
    }
}

7. Define the service interface and implementation (producer):

public interface DisruptorMqService {
    void sayHelloMq(String message);
}

@Slf4j
@Component
@Service
public class DisruptorMqServiceImpl implements DisruptorMqService {
    @Autowired
    private RingBuffer
messageModelRingBuffer;
    @Override
    public void sayHelloMq(String message) {
        log.info("record the message: {}", message);
        long sequence = messageModelRingBuffer.next();
        try {
            MessageModel event = messageModelRingBuffer.get(sequence);
            event.setMessage(message);
            log.info("Added message to queue: {}", event);
        } catch (Exception e) {
            log.error("failed to add event", e);
        } finally {
            messageModelRingBuffer.publish(sequence);
        }
    }
}

8. Write a test class to send a message:

@RunWith(SpringRunner.class)
@SpringBootTest(classes = DemoApplication.class)
public class DemoApplicationTests {
    @Autowired
    private DisruptorMqService disruptorMqService;
    @Test
    public void sayHelloMqTest() throws Exception {
        disruptorMqService.sayHelloMq("Message arrived, Hello world!");
        log.info("Message queue sent");
        Thread.sleep(2000);
    }
}

Test output shows the producer logging the message and the consumer processing it asynchronously.

Summary

The producer‑consumer pattern is common, but Disruptor implements it in‑memory with a lock‑free design, which explains its high efficiency and suitability for ultra‑low‑latency scenarios.