浅析Disruptor高性能线程消息传递并发框架

/**
 * Created by kl on 2018/8/24.
 * Content :进程日志事件内容载体
 */
public class LoggerEvent {
    private LoggerMessage log;
    public LoggerMessage getLog() {
        return log;
    }
    public void setLog(LoggerMessage log) {
        this.log = log;
    }
}

/**
 * Created by kl on 2018/8/24.
 * Content :进程日志事件工厂类
 */
public class LoggerEventFactory implements EventFactory{
    @Override
    public LoggerEvent newInstance() {
        return new LoggerEvent();
    }
}

/**
 * Created by kl on 2018/8/24.
 * Content :进程日志事件处理器
 */
@Component
public class LoggerEventHandler implements EventHandler{
    @Autowired
    private SimpMessagingTemplate messagingTemplate;
    @Override
    public void onEvent(LoggerEvent stringEvent, long l, boolean b) {
        messagingTemplate.convertAndSend("/topic/pullLogger",stringEvent.getLog());
    }
}

/**
 * Created by kl on 2018/8/24.
 * Content :Disruptor 环形队列
 */
@Component
public class LoggerDisruptorQueue {
    private Executor executor = Executors.newCachedThreadPool();
    // The factory for the event
    private LoggerEventFactory factory = new LoggerEventFactory();
    private FileLoggerEventFactory fileLoggerEventFactory = new FileLoggerEventFactory();
    // Specify the size of the ring buffer, must be power of 2.
    private int bufferSize = 2 * 1024;
    // Construct the Disruptor
    private Disruptordisruptor = new Disruptor<>(factory, bufferSize, executor);;
    private DisruptorfileLoggerEventDisruptor = new Disruptor<>(fileLoggerEventFactory, bufferSize, executor);;
    private static  RingBufferringBuffer;
    private static  RingBufferfileLoggerEventRingBuffer;
    @Autowired
    LoggerDisruptorQueue(LoggerEventHandler eventHandler,FileLoggerEventHandler fileLoggerEventHandler) {
        disruptor.handleEventsWith(eventHandler);
        fileLoggerEventDisruptor.handleEventsWith(fileLoggerEventHandler);
        this.ringBuffer = disruptor.getRingBuffer();
        this.fileLoggerEventRingBuffer = fileLoggerEventDisruptor.getRingBuffer();
        disruptor.start();
        fileLoggerEventDisruptor.start();
    }
    public static void publishEvent(LoggerMessage log) {
        long sequence = ringBuffer.next();  // Grab the next sequence
        try {
            LoggerEvent event = ringBuffer.get(sequence); // Get the entry in the Disruptor
            // for the sequence
            event.setLog(log);  // Fill with data
        } finally {
            ringBuffer.publish(sequence);
        }
    }
    public static void publishEvent(String log) {
        if(fileLoggerEventRingBuffer == null) return;
        long sequence = fileLoggerEventRingBuffer.next();  // Grab the next sequence
        try {
            FileLoggerEvent event = fileLoggerEventRingBuffer.get(sequence); // Get the entry in the Disruptor
            // for the sequence
            event.setLog(log);  // Fill with data
        } finally {
            fileLoggerEventRingBuffer.publish(sequence);
        }
    }
}