disruptor模拟高速处理大规模订单类业务场景

系统管理员 2023-10-18 15:12 65阅读 0赞

### **什么是Disruptor？** ###

Disruptor是一个开源的JAVA框架，它被设计用于在生产者—消费者（producer-consumer problem，简称PCP）问题上获得尽量高的吞吐量（TPS）和尽量低的延迟。Disruptor是LMAX在线交易平台的关键组成部分，LMAX平台使用该框架对订单处理速度能达到600万TPS，除金融领域之外，其他一般的应用中都可以用到Disruptor，它可以带来显著的性能提升。其实Disruptor与其说是一个框架，不如说是一种设计思路，这个设计思路对于存在“并发、缓冲区、生产者—消费者模型、事务处理”这些元素的程序来说，Disruptor提出了一种大幅提升性能（TPS）的方案。

> Disruptor是一个高性能的异步处理框架，或者可以认为是最快的消息框架（轻量的JMS），也可以认为是一个观察者模式的实现，或者事件监听模式的实现。

我们可以用下面的这几个图形象的理解一下该框架：  
![在这里插入图片描述][20190305192523696.png]

![在这里插入图片描述][20190305192603870.png]

![在这里插入图片描述][20190305192618132.png]

![在这里插入图片描述][20190305192629516.png]

简单来说，可以这样理解，有一个环形仓库，用于存放生产者生产出来的数据，生产者生成了数据马上放到环状上的某个节点【槽】上，然后生产者的指针指向下一个空槽，消费者呢，同样有一个指针，指向某个放置叾数据的槽不断从里面获取数据，这样整个过程就流通起来了，其实这个原理在redis的底层实现上也有类似的应用，有兴趣的伙伴可以去查查相关资料，

同时，Disruptor里面有一个很有用的操作，就是可以人为的涉及生产者和消费者的处理顺序，即线程执行顺序，也叫编排顺序，

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 1]

**用上面这两张图表示，我们可以使用Disruptor 设定不同消费者的执行顺序，这在一些复杂特殊的业务场景中很有必要，说到这儿是不是觉得很像大数据里面的一个实时数据处理框架storm的处理逻辑呢？**

基本上理论知识就普普及到这里了，下面我们来用Disruptor 做一个简单的业务逻辑处理，一个线程产生10000个订单，有3个消费者线程，一个消费者线程模拟订单入库逻辑，一个模拟发送消息给用户，这两个线程全部处理完毕再执行增值业务逻辑，模型图如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 2]

下面直接上代码，

<dependencies>
            
            <dependency>
                <groupId>com.lmax</groupId>
                <artifactId>disruptor</artifactId>
                <version>3.4.1</version>
            </dependency>
    
        </dependencies>

1、创建一个订单实体对象，作为流通在环形内存中的对象，

package com.congge.model;
    
    public class TradeTransaction {
    
        private String id;//交易订单号
        private double price;
    
        public TradeTransaction() {
        }
        public TradeTransaction(String id, double price) {
            super();
            this.id = id;
            this.price = price;
        }
        public String getId() {
            return id;
        }
        public void setId(String id) {
            this.id = id;
        }
        public double getPrice() {
            return price;
        }
        public void setPrice(double price) {
            this.price = price;
        }
    
    }

2、处理订单入库线程，

package com.congge.model;
    
    import com.lmax.disruptor.EventHandler;
    import com.lmax.disruptor.WorkHandler;
    
    import java.util.UUID;
    
    /**
     * 处理商品入库线程
     */
    public class TradeTransactionInDBHandler implements EventHandler<TradeTransaction>, WorkHandler<TradeTransaction> {
    
        public void onEvent(TradeTransaction tradeTransaction, long l, boolean b) throws Exception {
            this.onEvent(tradeTransaction);
        }
    
        public void onEvent(TradeTransaction event) throws Exception {
            String threadName = Thread.currentThread().getName();
            String id = event.getId();
            System.out.println("线程:" + threadName + "处理订单业务入库服务  ------> :  商品ID是 ：" + id);
        }
    }

3、处理发送消息线程，

package com.congge.model;
    
    import com.lmax.disruptor.EventHandler;
    
    /**
     * 处理订单发送短信业务
     */
    public class TradeTransactionJMSNotifyHandler implements EventHandler<TradeTransaction> {
    
        public void onEvent(TradeTransaction tradeTransaction, long l, boolean b) throws Exception {
            String threadName = Thread.currentThread().getName();
            String id = tradeTransaction.getId();
            System.out.println("线程:" + threadName + "处理订单业务发送短信通知服务:  商品ID是 ：" + id);
        }
    }

4、处理增值服务线程，

package com.congge.model;
    
    import com.lmax.disruptor.EventHandler;
    
    /**
     * 处理增值业务的线程
     */
    public class TradeTransactionVasHandler implements EventHandler<TradeTransaction> {
    
        public void onEvent(TradeTransaction tradeTransaction, long l, boolean b) throws Exception {
            String threadName = Thread.currentThread().getName();
            String id = tradeTransaction.getId();
            System.out.println("线程:" + threadName + "处理订单增值服务: 处理的 商品ID是 ：" + id);
    
        }
    }

5、生产者线程，模拟产生10000个订单，

package com.congge.model;
    
    import com.lmax.disruptor.EventTranslator;
    import com.lmax.disruptor.dsl.Disruptor;
    
    import java.util.Random;
    import java.util.concurrent.CountDownLatch;
    
    /**
     * 模拟生产者，产生订单线程
     */
    public class TradeTransactionPublisher implements Runnable {
    
        Disruptor<TradeTransaction> disruptor;
        private CountDownLatch latch;
        private static int LOOP = 10000;   //模拟一万次交易的发生
    
        public TradeTransactionPublisher(CountDownLatch latch,Disruptor<TradeTransaction> disruptor) {
            this.disruptor=disruptor;
            this.latch=latch;
        }
    
        public void run() {
            TradeTransactionEventTranslator tradeTransloator = new TradeTransactionEventTranslator();
            for(int i=0;i<LOOP;i++){
                disruptor.publishEvent(tradeTransloator);
            }
            latch.countDown();
        }
    }
    
    class TradeTransactionEventTranslator implements EventTranslator<TradeTransaction> {
    
        private Random random = new Random();
    
        public void translateTo(TradeTransaction event, long sequence) {
            this.generateTradeTransaction(event);
        }
    
        //生成订单流水号逻辑
        private TradeTransaction generateTradeTransaction(TradeTransaction trade) {
            trade.setId(random.nextInt() * 9999+"");
            return trade;
        }
    }

6、测试类：

package com.congge.model;
    
    import com.lmax.disruptor.BusySpinWaitStrategy;
    import com.lmax.disruptor.EventFactory;
    import com.lmax.disruptor.dsl.Disruptor;
    import com.lmax.disruptor.dsl.EventHandlerGroup;
    import com.lmax.disruptor.dsl.ProducerType;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class TestDemo3 {
    
        public static void main(String[] args) throws Exception {
    
            long beginTime=System.currentTimeMillis();
    
            int bufferSize=1024;
            ExecutorService executor= Executors.newFixedThreadPool(4);
    
            Disruptor<TradeTransaction> disruptor=new Disruptor<TradeTransaction>(new EventFactory<TradeTransaction>() {
                public TradeTransaction newInstance() {
                    return new TradeTransaction();
                }
            }, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());
    
            //使用disruptor创建消费者组C1,C2
            EventHandlerGroup<TradeTransaction> handlerGroup=disruptor.handleEventsWith(new TradeTransactionVasHandler(),new TradeTransactionInDBHandler());
    
            TradeTransactionJMSNotifyHandler jmsConsumer=new TradeTransactionJMSNotifyHandler();
            //声明在C1,C2之后执行JMS消息发送操作 也就是流程走到C3
            handlerGroup.then(jmsConsumer);
    
            disruptor.start();//启动
            CountDownLatch latch=new CountDownLatch(1);
    
            //生产者准备
            executor.submit(new TradeTransactionPublisher(latch, disruptor));
            latch.await();//等待生产者结束
            disruptor.shutdown();
            executor.shutdown();
            System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));
        }
    }

运行一下，让我们看一下效果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 3]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 4]

可以看到，这个过程的执行非常快，10秒多的时间全部处理完，当然，实际中，我们的每个业务逻辑处理线程里面还有大量的其他业务需要处理，但可以肯定的是，使用Disruptor来处理，可以充分使用CPU的性能，这里做一个小小的提示也是我们实际业务处理中可能面临的问题，就是当你的生产者线程和消费者线程处理逻辑的速度存在一定的时差怎么处理呢？可以考虑的方式是在生产者生产完毕数据后等待一小段时间，留给消费者处理，至于这个时间是多大，可以根据业务的处理时间进行粗略的计算。

**Disruptor 的简单使用就说到这里，大家有兴趣可以进行更深入的研究，谢谢观看！**

[20190305192523696.png]: https://img-blog.csdnimg.cn/20190305192523696.png
[20190305192603870.png]: https://img-blog.csdnimg.cn/20190305192603870.png
[20190305192618132.png]: https://img-blog.csdnimg.cn/20190305192618132.png
[20190305192629516.png]: https://img-blog.csdnimg.cn/20190305192629516.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2019030519313521.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20190305193155643.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20190305193648623.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20190305194313342.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20190305194325591.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw,size_16,color_FFFFFF,t_70