<!--Disruptor-->
<dependency>
    <groupId>com.lmax</groupId>
    <artifactId>disruptor</artifactId>
    <version>3.4.4</version>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>

package com.example.aopdemo.disruptor;

import lombok.Data;

/**
 * @author qx
 * @date 2024/2/21
 * @des 消息體
 */
@Data
public class MessageModel {

    private String message;

}

package com.example.aopdemo.disruptor;

import com.lmax.disruptor.EventFactory;

/**
 * @author qx
 * @date 2024/2/21
 * @des 事件工廠類
 */
public class MessageEventFactory implements EventFactory<MessageModel> {
    @Override
    public MessageModel newInstance() {
        return new MessageModel();
    }
}

package com.example.aopdemo.disruptor;

import com.lmax.disruptor.EventHandler;
import lombok.extern.slf4j.Slf4j;

/**
 * @author qx
 * @date 2024/2/21
 * @des 消息消費者
 */
@Slf4j
public class MessageEventHandler implements EventHandler<MessageModel> {
    @Override
    public void onEvent(MessageModel messageModel, long sequence, boolean endOfBatch) {
        log.info("消費者獲取消息:{}", messageModel);
    }
}

package com.example.aopdemo.disruptor;

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

/**
 * @author qx
 * @date 2024/2/21
 * @des
 */
@Component
public class BeanManager implements ApplicationContextAware {

    private static ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        BeanManager.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);
    }
}

package com.example.aopdemo.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author qx
 * @date 2024/2/21
 * @des 事件管理器
 */
@Configuration
public class MessageManager {

    @Bean("messageModel")
    public RingBuffer<MessageModel> messageModelRingBuffer() {
        // 定義線程池
        ExecutorService executorService = Executors.newFixedThreadPool(2);

        // 指定事件工廠
        MessageEventFactory factory = new MessageEventFactory();

        // 指定ringbuffer字節(jié)大小，必須為2的N次方（能將求模運算轉為位運算提高效率），否則將影響效率
        int bufferSize = 1024 * 256;

        //單線程模式，獲取額外的性能
        Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize, executorService, ProducerType.SINGLE, new BlockingWaitStrategy());

        //設置事件業(yè)務處理器---消費者
        disruptor.handleEventsWith(new MessageEventHandler());

        //啟動disruptor線程
        disruptor.start();

        //獲取ringbuffer環(huán)，用于接取生產者生產的事件
        return disruptor.getRingBuffer();
    }

}

package com.example.aopdemo.disruptor;

import com.lmax.disruptor.RingBuffer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @author qx
 * @date 2024/2/21
 * @des 生產者
 */
@Service
@Slf4j
public class DisruptorService {

    @Autowired
    private RingBuffer<MessageModel> messageModelRingBuffer;

    public void sayMessage(String message) {
        // 獲取下一個Event槽的下標
        long sequence = messageModelRingBuffer.next();
        try {
            // 填充數(shù)據(jù)
            MessageModel messageModel = messageModelRingBuffer.get(sequence);
            messageModel.setMessage(message);
            log.info("往消息隊列中添加消息:{}", messageModel);
        } 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);
        }

    }

}

package com.example.aopdemo.controller;

import com.example.aopdemo.disruptor.DisruptorService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author qx
 * @date 2024/2/21
 * @des Disruptor測試
 */
@RestController
public class DisruptorController {

    @Autowired
    private DisruptorService disruptorService;

    @GetMapping("/disruptor")
    public String disruptorTest(String message) {
        disruptorService.sayMessage(message);
        return "發(fā)送消息成功";
    }
}