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

/**
 * 消息體
 */
@Data
public class MessageModel {
    private String message;
}

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

@Slf4j
public class HelloEventHandler implements EventHandler<MessageModel> {
    @Override
    public void onEvent(MessageModel event, long sequence, boolean endOfBatch) {
        try {
            //這里停止1000ms是為了確定消費(fèi)消息是異步的
            Thread.sleep(1000);
            log.info("消費(fèi)者處理消息開始");
            if (event != null) {
                log.info("消費(fèi)者消費(fèi)的信息是：{}",event);
            }
        } catch (Exception e) {
            log.info("消費(fèi)者處理消息失敗");
        }
        log.info("消費(fèi)者處理消息結(jié)束");
    }
}

/**
 * 獲取實(shí)例化對象
 */
@Component
public class BeanManager implements ApplicationContextAware {
    private static ApplicationContext applicationContext = null;
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
    public static ApplicationContext getApplicationContext() { return applicationContext; }
    public static Object getBean(String name) {
        return applicationContext.getBean(name);
    }
    public static <T> T getBean(Class<T> clazz) {
        return applicationContext.getBean(clazz);
    }
}

@Configuration
public class MQManager {
    @Bean("messageModel")
    public RingBuffer<MessageModel> messageModelRingBuffer() {
        //定義用于事件處理的線程池， Disruptor通過java.util.concurrent.ExecutorSerivce提供的線程來觸發(fā)consumer的事件處理
        ExecutorService executor = Executors.newFixedThreadPool(2);
        //指定事件工廠
        HelloEventFactory factory = new HelloEventFactory();
        //指定ringbuffer字節(jié)大小，必須為2的N次方（能將求模運(yùn)算轉(zhuǎn)為位運(yùn)算提高效率），否則將影響效率
        int bufferSize = 1024 * 256;
        //單線程模式，獲取額外的性能
        Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize, executor,
                ProducerType.SINGLE, new BlockingWaitStrategy());
        //設(shè)置事件業(yè)務(wù)處理器---消費(fèi)者
        disruptor.handleEventsWith(new HelloEventHandler());
        // 啟動disruptor線程
        disruptor.start();
        //獲取ringbuffer環(huán)，用于接取生產(chǎn)者生產(chǎn)的事件
        RingBuffer<MessageModel> ringBuffer = disruptor.getRingBuffer();
        return ringBuffer;
    }
}

public interface DisruptorMqService {
    /**
     * 消息
     * @param message
     */
    void sayHelloMq(String message);
}
@Slf4j
@Component
@Service
public class DisruptorMqServiceImpl implements DisruptorMqService {
    @Autowired
    private RingBuffer<MessageModel> 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è)請求的sequence未被提交將會堵塞后續(xù)的發(fā)布操作或者其他的producer
            messageModelRingBuffer.publish(sequence);
        }
    }
}

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = DemoApplication.class)
public class DemoApplicationTests {
    @Autowired
    private DisruptorMqService disruptorMqService;
    /**
     * 項(xiàng)目內(nèi)部使用Disruptor做消息隊(duì)列
     * @throws Exception
     */
    @Test
    public void sayHelloMqTest() throws Exception{
        disruptorMqService.sayHelloMq("消息到了，Hello world!");
        log.info("消息隊(duì)列已發(fā)送完畢");
        //這里停止2000ms是為了確定是處理消息是異步的
        Thread.sleep(2000);
    }
}