内存中的消息队列-disruptor-526互联

一、介绍
工作中遇到项目使用Disruptor做消息队列,对你没看错,不是Kafka,也不是rabbitmq;Disruptor有个最大的优点就是快,还有一点它是开源的。
Disruptor 是英国外汇交易公司LMAX开发的一个高性能队列。

Disruptor是一个开源的Java框架，它被设计用于在生产者—消费者（producer-consumer problem，简称PCP）问题上获得尽量高的吞吐量（TPS）和尽量低的延迟。
从功能上来看，Disruptor 是实现了“队列”的功能，而且是一个有界队列。那么它的应用场景自然就是“生产者-消费者”模型的应用场合了。
Disruptor是LMAX在线交易平台的关键组成部分，LMAX平台使用该框架对订单处理速度能达到600万TPS，除金融领域之外，其他一般的应用中都可以用到Disruptor，它可以带来显著的性能提升。
其实Disruptor与其说是一个框架，不如说是一种设计思路，这个设计思路对于存在“并发、缓冲区、生产者—消费者模型、事务处理”这些元素的程序来说，Disruptor提出了一种大幅提升性能（TPS）的方案。
Disruptor的github主页：https://github.com/LMAX-Exchange/disruptor
Disruptor 定义的事件处理接口，由用户实现，用于处理事件，是 Consumer 的真正实现
Event 在 Disruptor 的语义中，生产者和消费者之间进行交换的数据被称为事件(Event)。它不是一个被 Disruptor 定义的特定类型，而是由 Disruptor 的使用者定义并指定。
EventHandler Disruptor 定义的事件处理接口，由用户实现，用于处理事件，是 Consumer 的真正实现
Sequencer 是 Disruptor 的真正核心。此接口有两个实现类 SingleProducerSequencer、MultiProducerSequencer ，它们定义在生产者和消费者之间快速、正确地传递数据的并发算法。
RingBuffer 如其名，环形的缓冲区。曾经 RingBuffer 是 Disruptor 中的最主要的对象，但从3.0版本开始，其职责被简化为仅仅负责对通过 Disruptor 进行交换的数据（事件）进行存储和更新。在一些更高级的应用场景中，Ring Buffer 可以由用户的自定义实现来完全替代。
Sequence Disruptor 通过顺序递增的序号来编号管理通过其进行交换的数据（事件），对数据(事件)的处理过程总是沿着序号逐个递增处理。一个 Sequence 用于跟踪标识某个特定的事件处理者( RingBuffer/Consumer )的处理进度。虽然一个 AtomicLong 也可以用于标识进度，但定义 Sequence 来负责该问题还有另一个目的，那就是防止不同的 Sequence 之间的CPU缓存伪共享(Flase Sharing)问题。
Sequence Barrier 用于保持对RingBuffer的 main published Sequence 和Consumer依赖的其它Consumer的 Sequence 的引用。Sequence Barrier 还定义了决定 Consumer 是否还有可处理的事件的逻辑。
Wait Strategy 定义 Consumer 如何进行等待下一个事件的策略。（注：Disruptor 定义了多种不同的策略，针对不同的场景，提供了不一样的性能表现）
Producer 即生产者，只是泛指调用 Disruptor 发布事件的用户代码，Disruptor 没有定义特定接口或类型。

二、实战操作


1、引入pom依赖

<!-- disruptor 高性能消息队列 -->
<dependency>
    <groupId>com.lmax</groupId>
    <artifactId>disruptor</artifactId>
    <version>3.4.4</version>
</dependency>

2、构建消息体Model实体

@Data
public class MessageModel {

    /**
     * 消息类型
     */
    private int type;

    /**
     * 消息内容
     */
    private Object data；
}
3、构造EventFactory 消息工厂

public class MessageEventFactory implements EventFactory<MessageModel> {

    @Override
    public MessageModel newInstance() {
        return new MessageModel();
    }
}
4、构造EventHandler-消费者

@Slf4j
@Component
public class MessageEventHandler implements EventHandler<MessageModel> {
    // 自行添加消息模板
    @Autowired
    private MessageTemplateService messageTemplateService;

    @Override
    public void onEvent(MessageModel event, long sequence, boolean endOfBatch) {
        try {
            if (event != null) {
                switch (event.getType()) {
                    case 1:
                        // 站内消息/邮件/短信通知发送
                        log.info("站内消息/邮件/短信通知发送");
                        messageTemplateService.sendMessage((SendMessageDTO) event.getData());
                        break;
                    default:
                        break;
                }
            }
        } catch (Exception e) {
            log.info("消费者处理消息失败,{}",e);
        }
    }
}
5、构造BeanManager  来获取实例化对象

/**
 * ApplicationContextHelper
 */
@Component
public class ApplicationContext implements ApplicationContextAware {
    /**
     * 上下文对象实例
     */
    private static org.springframework.context.ApplicationContext applicationContext = null;

    @Override
    public void setApplicationContext(org.springframework.context.ApplicationContext applicationContext) throws BeansException {
        if(ApplicationContext.applicationContext == null){
            ApplicationContext.applicationContext  = applicationContext;
        }
    }

    /**
     * 获取applicationContext
     * @return
     */
    public static org.springframework.context.ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    /**
     * 通过name获取 Bean.
     * @param name
     * @return
     */
    public static Object getBean(String name){
        org.springframework.context.ApplicationContext applicationContext = getApplicationContext();
        if(applicationContext == null){
            return null;
        }
        return applicationContext.getBean(name);
    }

    /**
     * 通过class获取Bean.
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T getBean(Class<T> clazz){
        org.springframework.context.ApplicationContext applicationContext = getApplicationContext();
        if(applicationContext == null){
            return null;
        }
        return applicationContext.getBean(clazz);
    }

    /**
     * 通过name,以及Clazz返回指定的Bean
     * @param name
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T getBean(String name,Class<T> clazz){
        org.springframework.context.ApplicationContext applicationContext = getApplicationContext();
        if(applicationContext == null){
            return null;
        }
        return applicationContext.getBean(name, clazz);
    }
}
6、构造MQManager

/**
 * 构建消息实例对象
 * 
 */
@Configuration
public class MessageManager {

    @Autowired
    private MessageEventHandler messageEventHandler;

    @Bean("messageModel")
    public RingBuffer<MessageModel> messageModelRingBuffer() {
        //指定事件工厂
        MessageEventFactory factory = new MessageEventFactory();
        //指定ringbuffer字节大小，必须为2的N次方（能将求模运算转为位运算提高效率），否则将影响效率
        int bufferSize = 1024 * 256;
        //单线程模式，获取额外的性能
        Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize, ThreadFactory.create("mq"));
        //设置事件业务处理器---消费者
        disruptor.handleEventsWith(messageEventHandler);
        // 启动disruptor线程
        disruptor.start();
        //获取ringbuffer环，用于接取生产者生产的事件
        RingBuffer<MessageModel> ringBuffer = disruptor.getRingBuffer();
        return ringBuffer;
    }
}
// 设置线程池

/**
 * 封装线程名称
 */
public class ThreadFactory implements java.util.concurrent.ThreadFactory {

    public static ThreadFactory create(String namePrefix) {
        return new ThreadFactory(namePrefix);
    }

    private final AtomicInteger poolNumber = new AtomicInteger(1);

    private final ThreadGroup group;

    private final AtomicInteger threadNumber = new AtomicInteger(1);

    private final String namePrefix;

    private ThreadFactory(String namePrefix) {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
        this.namePrefix = namePrefix + " pool " + poolNumber.getAndIncrement() + "-thread-";
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
        if (t.isDaemon()) {
            t.setDaemon(false);
        }

        if (t.getPriority() != Thread.NORM_PRIORITY) {
            t.setPriority(Thread.NORM_PRIORITY);
        }
        return t;
    }
}

7、构造Mqservice和实现类-生产者

/**
 * 消息发送服务
 *
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DisruptorMqService {

    private final RingBuffer<MessageModel> messageModelRingBuffer;

    public void sendMessage(int type,Object data) {
        log.info("record the message: {}", type);
        //获取下一个Event槽的下标
        long sequence = messageModelRingBuffer.next();
        try {
            //给Event填充数据
            MessageModel event = messageModelRingBuffer.get(sequence);
            event.setType(type);
            event.setData(data);
            log.info("往消息队列中添加消息：{}", event);
        } catch (Exception e) {
            log.error("failed to add event to messageModelRingBuffer for : e = {},{}", e, e.getMessage());
        } finally {
            //发布Event，激活观察者去消费，将sequence传递给改消费者
            //注意最后的publish方法必须放在finally中以确保必须得到调用；如果某个请求的sequence未被提交将会堵塞后续的发布操作或者其他的producer
            messageModelRingBuffer.publish(sequence);
        }
    }
}

8、测试
// 引入服务类

private final DisruptorMqService disruptorMqService;
//调用sendMessage()方法

disruptorMqService.sendMessage(1,new SendMessageDTO().setCode(MessageTemplateEnum.FIN_OFFLINE_DOCK.getValue())
        .setParams(params)
        .setOrgId(dockInfoEntity.getEnterpriseId()));

总结

其实生成者 -> 消费者模式是很常见的，通过一些消息队列也可以轻松做到上述的效果。不同的地方在于，Disruptor 是在内存中以队列的方式去实现的，而且是无锁的。这也是 Disruptor 为什么高效的原因。