Introduction and Usage of the Disruptor Framework

🌐 中文

English

Introduction

When developing multi-user team collaboration or real-time chat systems using WebSocket, a common performance bottleneck arises: when a WebSocket thread receives a team message, if it synchronously handles time-consuming business logic (such as database persistence, permission validation, broadcast distribution, etc.) directly within the current thread, that thread becomes occupied for an extended period.

In high-concurrency scenarios, this synchronous processing approach can quickly exhaust the network thread pool, causing subsequent team messages to be blocked, potentially leading to connection timeouts, message loss, and a dramatic drop in overall system throughput.

To solve this problem, we need to introduce an asynchronous decoupling mechanism. Among various queueing solutions, the LMAX Disruptor framework, with its exceptional performance, has become our preferred tool for achieving efficient, low-latency team message processing.

Introduction to Disruptor

Disruptor is a high-performance, low-latency in-memory concurrent framework developed by the British foreign exchange trading company LMAX. Its original design goal was to address the low-latency and high-throughput requirements of financial trading systems in multi-threaded environments.

In traditional Java concurrent programming, we typically use ArrayBlockingQueue or LinkedBlockingQueue to implement the producer-consumer model. However, Disruptor takes a different path, abandoning the traditional queue structure and achieving astonishing performance of processing millions of orders per second with almost no explicit locks (Lock-Free).

1. Performance Pain Points of Traditional Queues

Before understanding why Disruptor is fast, we need to know the three major pain points of traditional concurrent queues (like BlockingQueue) under high concurrency:

1. Severe Lock Contention: Traditional blocking queues typically rely on heavyweight locks like ReentrantLock or synchronized to ensure thread safety during enqueue (Put) and dequeue (Take) operations. Under high concurrency, the overhead of thread context switching is enormous.
2. False Sharing Problem: In traditional linked list or array structures, the queue’s head and tail pointers are often located close together in memory. When multi-core CPUs read them, they are loaded into the same cache line, causing frequent cache invalidation between cores, triggering writes back to main memory, and severely slowing down CPU efficiency.
3. Garbage Collection (GC) Pressure: Frequent enqueue and dequeue operations involve the creation and destruction of many objects, easily causing the JVM to trigger frequent GC, leading to system pauses (Stop-The-World).

2. Disruptor’s Core Black Magic

To squeeze the last drop of performance from the CPU, Disruptor introduces the following core designs:

① Ring Buffer Structure (RingBuffer)

The core of Disruptor uses a fixed-size ring array called RingBuffer.

• No GC Pressure: The array is pre-filled with empty data objects (Events) upon initialization. During runtime, producers only modify the fields of these pre-existing objects rather than creating new ones. This achieves memory reuse, generating almost no new GC.
• Bitwise Operation Addressing: By setting the array size to a power of two (e.g., 1024), finding the next available position can use bitwise operations (sequence & (length - 1)) instead of modulo operations, which is extremely efficient.

② Innovative Lock-Free Design: Sequence

Disruptor does not use traditional head/tail pointers. Instead, each producer and consumer maintains its own Sequence (a sequence number, essentially an incrementing long value).

• Threads determine if there is new data to read or space to write by comparing their respective Sequence values.
• Internally, it uses underlying memory barriers to prevent CPU instruction reordering and CAS (Compare And Swap) atomic operations to replace traditional explicit locks, achieving complete lock-free operation.

③ Solving False Sharing: Cache Line Padding

In the underlying implementation of core components (like Sequence), Disruptor artificially pads the core variable with 7 unused long variables before and after it. This ensures the core sequence number occupies an entire 64-byte CPU cache line exclusively, completely avoiding the false sharing problem and maximizing cache hit rates for multi-core CPUs.

How to Import the Disruptor Framework in SpringBoot

Add the disruptor dependency in pom.xml:

1
2
3
4
5
<dependency>
    <groupId>com.lmax</groupId>
    <artifactId>disruptor</artifactId>
    <version>3.4.2</version>
</dependency>

Basic Usage Example of Disruptor

Below is a simple example demonstrating how to use Disruptor to implement a producer-consumer model:

1. Define the Event

The Event is the data carrier flowing within the RingBuffer. Disruptor pre-instantiates these Events upon startup.

1
2
3
4
5
6
7
@Data
public class TeamMessageEvent {
    private String messagePayload;     // Message content (JSON string or text)
    private WebSocketSession session;  // Sender's WebSocket session
    private String teamId;             // Team/Room ID
    private String userId;             // User ID
}

1. Define the Consumer (WorkHandler)

The consumer is responsible for fetching data from the RingBuffer and executing the specific time-consuming business logic. Here we use WorkHandler, which supports multiple consumer threads working together, where a single message is processed by only one of the consumers (work queue pattern).

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@Component
@Slf4j
public class TeamMessageWorkHandler implements WorkHandler<TeamMessageEvent> {

    @Override
    public void onEvent(TeamMessageEvent event) throws Exception {
        // 1. Extract data from the event
        String payload = event.getMessagePayload();
        WebSocketSession session = event.getSession();
        String teamId = event.getTeamId();
        
        log.info("线程 {} 开始异步处理团队 [{}] 的消息: {}", 
                 Thread.currentThread().getName(), teamId, payload);

        // 2. Simulate time-consuming business logic (e.g., database persistence, complex business calculations)
        Thread.sleep(100); 

        // 3. Construct a response and broadcast it to other team members via WebSocket
        String response = "{\"status\":\"success\",\"data\":\"处理完成\"}";
        if (session != null && session.isOpen()) {
            session.sendMessage(new TextMessage(response));
        }
    }
}

1. Configure the Disruptor Instance

Configure Disruptor in Spring Boot. Set the RingBuffer size appropriately and bind the consumer thread pool.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@Configuration
public class DisruptorConfig {

    @Resource
    private TeamMessageWorkHandler teamMessageWorkHandler;

    @Bean("teamMessageDisruptor")
    public Disruptor<TeamMessageEvent> teamMessageDisruptor() {
        // Specify RingBuffer size, must be a power of two
        int bufferSize = 1024 * 256;

        // Create Disruptor
        Disruptor<TeamMessageEvent> disruptor = new Disruptor<>(
                TeamMessageEvent::new,
                bufferSize,
                ThreadFactoryBuilder.create()
                .setNamePrefix("disruptorWorker")
                .build()
        );

        // Set consumer handlers (running in WorkerPool mode, multiple threads collaborate on consumption)
        disruptor.handleEventsWithWorkerPool(teamMessageWorkHandler);

        // Start Disruptor
        disruptor.start();
        return disruptor;
    }
}

1. Define the Producer

The producer is responsible for delivering high-concurrency messages received by WebSocket into the Disruptor’s RingBuffer. The delivery process is lock-free and extremely fast.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@Component
public class TeamMessageProducer {

    @Resource
    private Disruptor<TeamMessageEvent> teamMessageDisruptor;

    /**
     * Publish a message to the RingBuffer
     */
    public void publishEvent(String payload, WebSocketSession session, String teamId, String userId) {
        RingBuffer<TeamMessageEvent> ringBuffer = teamMessageDisruptor.getRingBuffer();
        
        // 1. Get the next available sequence number (Sequence)
        long next = ringBuffer.next();
        try {
            // 2. Get the pre-allocated Event object corresponding to this sequence and populate its attributes
            TeamMessageEvent event = ringBuffer.get(next);
            event.setMessagePayload(payload);
            event.setSession(session);
            event.setTeamId(teamId);
            event.setUserId(userId);
        } finally {
            // 3. Publish the event (MUST be published in finally block to ensure sequence number is committed even if an exception occurs)
            ringBuffer.publish(next);
        }
    }

    /**
     * Graceful shutdown: Wait for all events to be processed before closing
     */
    @PreDestroy
    public void destroy() {
        if (teamMessageDisruptor != null) {
            teamMessageDisruptor.shutdown();
        }
    }
}

1. Integrate into the WebSocket Handler

Finally, in the WebSocket network event handler, we no longer process business logic synchronously. Instead, we immediately hand off the message to the Producer, thereby releasing the WebSocket thread instantly to handle the next network request.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Component
@Slf4j
public class GameTeamWebSocketHandler extends TextWebSocketHandler {

    @Resource
    private TeamMessageProducer teamMessageProducer;

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        // Extract common parameters established during handshake interception from session attributes
        String teamId = (String) session.getAttributes().get("teamId");
        String userId = (String) session.getAttributes().get("userId");
        String payload = message.getPayload();

        // Instantly "deliver" the high-concurrency network message to the asynchronous queue, performing no time-consuming processing here
        teamMessageProducer.publishEvent(payload, session, teamId, userId);
    }
}

FAQ

Why Does Disruptor Need Graceful Shutdown?

In production environments, distributed systems constantly face version updates, service restarts, or scaling down. If we forcefully terminate the process when shutting down the Spring container, it could lead to the following catastrophic consequences:

1. Data Loss: There might be backlogged team messages in the RingBuffer that haven’t been processed by consumers yet. A forceful shutdown would cause this data to vanish.

2. State Inconsistency: Consumer threads might be in a critical state mid-execution (e.g., just finished writing to table A in the database, but haven’t updated table B yet). Abrupt interruption could lead to dirty data states.

Summary

By introducing the Disruptor framework, we have successfully achieved full asynchronous decoupling between WebSocket network I/O threads and time-consuming business logic threads.

In high-concurrency environments, WebSocket threads only need to handle two extremely fast operations: “receive message -> insert into RingBuffer”. The single-event processing time plummets from hundreds of milliseconds to the nanosecond level. The underlying lock-free ring buffer and memory barrier black magic further ensure safety and high performance during multi-threaded message delivery, perfectly resolving the message blocking challenge in multi-user team collaboration systems.