<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是為了確定消費消息是異步的
            Thread.sleep(1000);
            log.info("消費者處理消息開始");
            if (event != null) {
                log.info("消費者消費的信息是：{}",event);
            }
        } catch (Exception e) {
            log.info("消費者處理消息失敗");
        }
        log.info("消費者處理消息結束");
    }
}

/**
?* 獲取實例化對象
?*/
@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次方（能將求模運算轉為位運算提高效率），否則將影響效率
? ? ? ? int bufferSize = 1024 * 256;
? ? ? ? //單線程模式，獲取額外的性能
? ? ? ? Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize, executor,
? ? ? ? ? ? ? ? ProducerType.SINGLE, new BlockingWaitStrategy());
? ? ? ? //設置事件業(yè)務處理器---消費者
? ? ? ? disruptor.handleEventsWith(new HelloEventHandler());
? ? ? ? // 啟動disruptor線程
? ? ? ? disruptor.start();
? ? ? ? //獲取ringbuffer環(huá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);
? ? ? ? //獲取下一個Event槽的下標
? ? ? ? long sequence = messageModelRingBuffer.next();
? ? ? ? try {
? ? ? ? ? ? //給Event填充數據
? ? ? ? ? ? MessageModel event = messageModelRingBuffer.get(sequence);
? ? ? ? ? ? event.setMessage(message);
? ? ? ? ? ? log.info("往消息隊列中添加消息：{}", event);
? ? ? ? } catch (Exception e) {
? ? ? ? ? ? log.error("failed to add event to messageModelRingBuffer for : e = {},{}",e,e.getMessage());
? ? ? ? } finally {
? ? ? ? ? ? //發(fā)布Event，激活觀察者去消費，將sequence傳遞給改消費者
? ? ? ? ? ? //注意最后的publish方法必須放在finally中以確保必須得到調用；如果某個請求的sequence未被提交將會堵塞后續(xù)的發(fā)布操作或者其他的producer
? ? ? ? ? ? messageModelRingBuffer.publish(sequence);
? ? ? ? }
? ? }
}

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