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

implementation 'com.lmax:disruptor:3.4.4'

@Data
public class LogEvent {
    private String message;
}

public class LogEventFactory implements EventFactory<LogEvent> {
    @Override
    public LogEvent newInstance() {
        return new LogEvent();
    }
}

public class LogEventHandler implements EventHandler<LogEvent> {
    @Override
    public void onEvent(LogEvent logEvent, long sequence, boolean endOfBatch) throws Exception {
        System.out.println(logEvent.getMessage());
    }
}

private static Disruptor<LogEvent> getLogEventDisruptor() {
    // 創(chuàng)建 LogEvent 的工廠
    LogEventFactory logEventFactory = new LogEventFactory();
    // Disruptor 的 RingBuffer 緩存大小
    int bufferSize = 1024 * 1024;
    // 生產(chǎn)者的線程工廠
    ThreadFactory threadFactory = new ThreadFactory() {
        final AtomicInteger threadNum = new AtomicInteger(0);
        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "LogEventThread" + " [#" + threadNum.incrementAndGet() + "]");
        }
    };
    //實(shí)例化 Disruptor
    return new Disruptor<>(
            logEventFactory,
            bufferSize,
            threadFactory,
            // 單生產(chǎn)者
            ProducerType.SINGLE,
            // 阻塞等待策略
            new BlockingWaitStrategy());
}

@Configuration
public class MQManager {
    @Bean("messageModel")
    public RingBuffer<LogEvent> messageModelRingBuffer() {
        //定義用于事件處理的線程池， Disruptor通過(guò)java.util.concurrent.ExecutorSerivce提供的線程來(lái)觸發(fā)consumer的事件處理
        // 生產(chǎn)者的線程工廠
    ThreadFactory threadFactory = new ThreadFactory() {
        final AtomicInteger threadNum = new AtomicInteger(0);
        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "LogEventThread" + " [#" + threadNum.incrementAndGet() + "]");
        }
    };
        //指定事件工廠
        LogEventFactory factory = new LogEventFactory();
        //指定ringbuffer字節(jié)大小，必須為2的N次方（能將求模運(yùn)算轉(zhuǎn)為位運(yùn)算提高效率），否則將影響效率
        int bufferSize = 1024 * 256;
        //單線程模式，獲取額外的性能
        Disruptor<LogEvent> disruptor = new Disruptor<>(factory,
             bufferSize, 
             threadFactory,
             ProducerType.SINGLE, 
             new BlockingWaitStrategy());
        //設(shè)置事件業(yè)務(wù)處理器---消費(fèi)者
        //Disruptor 的 handleEventsWith 方法來(lái)綁定處理事件的 Handler 對(duì)象。
        disruptor.handleEventsWith(new LogEventHandler ());
      // Disruptor 可以設(shè)置多個(gè)處理事件的 Handler，并且可以靈活的設(shè)置消費(fèi)者的處理順序，串行，并行都是可以的。
       //就比如下面的代碼表示 Handler1 和 Handler2 是并行執(zhí)行，最后再執(zhí)行 Handler3 。
       //disruptor.handleEventsWith(new Handler1(), new Handler2()).handleEventsWith(new Handler3());
        // 啟動(dòng)disruptor線程
        disruptor.start();
        //獲取ringbuffer環(huán)，用于接取生產(chǎn)者生產(chǎn)的事件
        RingBuffer<LogEvent> ringBuffer = disruptor.getRingBuffer();
        return ringBuffer;
    }

public class Disruptor<T> {
  public Disruptor(
          final EventFactory<T> eventFactory,
          final int ringBufferSize,
          final ThreadFactory threadFactory,
          final ProducerType producerType,
          final WaitStrategy waitStrategy)
  {
      this(
          RingBuffer.create(producerType, eventFactory, ringBufferSize, waitStrategy),
          new BasicExecutor(threadFactory));
  }
......
}

public interface WaitStrategy
{
    long waitFor(long sequence, Sequence cursor, Sequence dependentSequence, SequenceBarrier barrier)
        throws AlertException, InterruptedException, TimeoutException;
    void signalAllWhenBlocking();
}

public Disruptor(final EventFactory<T> eventFactory, final int ringBufferSize, final ThreadFactory threadFactory)
{
    this(RingBuffer.createMultiProducer(eventFactory, ringBufferSize), new BasicExecutor(threadFactory));
}

//獲取 Disruptor 對(duì)象
Disruptor<LogEvent> disruptor = getLogEventDisruptor();
//綁定處理事件的Handler對(duì)象
disruptor.handleEventsWith(new LogEventHandler());
//啟動(dòng) Disruptor
disruptor.start();
//獲取保存事件的環(huán)形數(shù)組（RingBuffer）
RingBuffer<LogEvent> ringBuffer = disruptor.getRingBuffer();
//發(fā)布 10w 個(gè)事件
for (int i = 1; i <= 100000; i++) {
    // 通過(guò)調(diào)用 RingBuffer 的 next() 方法獲取下一個(gè)空閑事件槽的序號(hào)
    long sequence = ringBuffer.next();
    try {
        LogEvent logEvent = ringBuffer.get(sequence);
        // 初始化 Event，對(duì)其賦值
        logEvent.setMessage("這是第%d條日志消息".formatted(i));
    } finally {
        // 發(fā)布事件
        ringBuffer.publish(sequence);
    }
}
// 關(guān)閉 Disruptor
disruptor.shutdown();

public interface DisruptorMqService {
    /**
     * 消息
     * @param message
     */
    void sayHelloMq(String message);
}
@Slf4j
@Component
@Service
public class DisruptorMqServiceImpl implements DisruptorMqService {
    @Autowired
    private RingBuffer<LogEvent> messageModelRingBuffer;
    @Override
    public void sayHelloMq(String message) {
        log.info("record the message: {}",message);
        //獲取下一個(gè)Event槽的下標(biāo)
        long sequence = messageModelRingBuffer.next();
        try {
            //給Event填充數(shù)據(jù)
            MessageModel event = messageModelRingBuffer.get(sequence);
            event.setMessage(message);
            log.info("往消息隊(duì)列中添加消息：{}", event);
        } catch (Exception e) {
            log.error("failed to add event to messageModelRingBuffer for : e = {},{}",e,e.getMessage());
        } finally {
            //發(fā)布Event，激活觀察者去消費(fèi)，將sequence傳遞給改消費(fèi)者
            //注意最后的publish方法必須放在finally中以確保必須得到調(diào)用；如果某個(gè)請(qǐng)求的sequence未被提交將會(huì)堵塞后續(xù)的發(fā)布操作或者其他的producer
            messageModelRingBuffer.publish(sequence);
        }
    }
}