java多线程之并发框架Disruptor

本是古典何须时尚 2024-04-17 19:17 27阅读 0赞

# 什么是Disruptor #

Disruptor是一个高性能的无锁线程间通讯框架，作者英国LMAX公司，其用于自己一种新型零售金融交易平台，它能够以很低的延迟产生大量交易，这个系统是建立在JVM平台上，其核心是一个业务逻辑处理器，它能够在一个线程里每秒处理6百万订单。业务逻辑处理器完全是运行在内存中，使用事件源驱动方式。业务逻辑处理器的核心是Disruptor。其最大的特点就是能够在无锁的情况下实现网络的Queue并发操作。

# Disruptor的特性 #

上面提到了Disruptor是一个**无锁的并发框架**，那么在多线程情况下如何避免线程间竞争的呢？答案就是在需要确保操作是线程安全的地方Disruptor使用CAS（Compare And Swap/Set）操作，这是一个CPU级别的指令，工作方式类似于乐观锁，跟新值的时候如果想要改变的值不再是原来的值，操作失败（有其他操作已经改变了这个值），否则操作成功。  
Disruptor在器数据结构上也非常特别使用了RingBuffer（环形缓冲区）来实现，是一种用于表示一个固定尺寸、头尾相连的缓冲区的数据结构，适合缓存数据流。RingBuffer 通常采用数组实现，对 CPU 缓存友好，性能比链表好。  
![在这里插入图片描述][20190830151046218.png]  
Disruptor的高性能还表现在其避免GC上，其利用环形缓冲区提前分配缓冲区元素类型对象，Ring Buffer会首先将整个缓冲区填满为Factory所产生的实例，新添加时只需要设置其中的值即可。

# Disruptor入门例子 #

Disruptor是通过事件源通知方式进行消息通讯，使用Disruptor需要知道三个重要的角色：事件、生产者和消费者，生产者和消费者均可以一个或者多个。有这样一个例子：在一个停车场的入口，每一辆车辆进来需要做三步处理，处理完后下一辆车才能进来，这三步分别是车辆数据入库、车辆数据进入Kafka消息队列、最后发送短信通知车主。通过分析我们可以知道车辆停车动作是事件，三步操作是消费者，而车辆则由生产者产生。

package com.luke.study.disruptor.event;
    
    /** * @Descrtption 事件（消息） * @Author luke * @Date 2019/8/29 **/
    public class InParkingDataEvent { 
    
        private String carLicense;
    
        public String getCarLicense() { 
            return carLicense;
        }
    
        public void setCarLicense(String carLicense) { 
            this.carLicense = carLicense;
        }
    
    }

package com.luke.study.disruptor.handler;
    
    import com.lmax.disruptor.EventHandler;
    import com.lmax.disruptor.WorkHandler;
    import com.luke.study.disruptor.event.InParkingDataEvent;
    
    /** * @Descrtption ParkingDataInDbHandler 车辆数据存入数据库处理器 * @Author luke * @Date 2019/8/29 **/
    public class ParkingDataInDbHandler implements EventHandler<InParkingDataEvent>, WorkHandler<InParkingDataEvent> { 
    
        @Override
        public void onEvent(InParkingDataEvent event) throws Exception { 
            long threadId = Thread.currentThread().getId();
            String carLicense = event.getCarLicense();
            System.out.println(String.format("Thread Id %s save %s into db ....",threadId,carLicense));
        }
    
        @Override
        public void onEvent(InParkingDataEvent event, long sequence, boolean endOfBatch) throws Exception { 
            this.onEvent(event);
        }
    
    }

package com.luke.study.disruptor.handler;
    
    import com.lmax.disruptor.EventHandler;
    import com.luke.study.disruptor.event.InParkingDataEvent;
    
    /** * @Descrtption ParkingDataInSmsHandler 车辆数据发送到kafka处理器 * @Author luke * @Date 2019/8/29 **/
    public class ParkingDataInKafkaHandler implements EventHandler<InParkingDataEvent> { 
    
        @Override
        public void onEvent(InParkingDataEvent event, long sequence, boolean endOfBatch) throws Exception { 
            long threadId = Thread.currentThread().getId();
            String carLicense = event.getCarLicense();
            System.out.println(String.format("Thread Id %s send %s in plaza sms to kafka",threadId,carLicense));
        }
    
    }

package com.luke.study.disruptor.handler;
    
    import com.lmax.disruptor.EventHandler;
    import com.luke.study.disruptor.event.InParkingDataEvent;
    
    /** * @Descrtption ParkingDataInSmsHandler 车辆数据发送短信处理器 * @Author luke * @Date 2019/8/29 **/
    public class ParkingDataInSmsHandler implements EventHandler<InParkingDataEvent> { 
    
        @Override
        public void onEvent(InParkingDataEvent event, long sequence, boolean endOfBatch) throws Exception { 
            long threadId = Thread.currentThread().getId();
            String carLicense = event.getCarLicense();
            System.out.println(String.format("Thread Id %s send %s in plaza sms to user",threadId,carLicense));
        }
    
    }

package com.luke.study.disruptor.publisher;
    
    import com.lmax.disruptor.EventTranslator;
    import com.lmax.disruptor.dsl.Disruptor;
    import com.luke.study.disruptor.event.InParkingDataEvent;
    import java.util.concurrent.CountDownLatch;
    
    /** * @Descrtption Publisher 事件发布类(生产者) * @Author luke * @Date 2019/8/29 **/
    public class InParkingDataEventPublisher implements Runnable{ 
    
        private Disruptor<InParkingDataEvent> disruptor;
    
        private CountDownLatch latch;
    
        public InParkingDataEventPublisher(Disruptor<InParkingDataEvent> disruptor, CountDownLatch latch) { 
            this.disruptor = disruptor;
            this.latch = latch;
        }
    
        /**模拟10辆车入场*/
        private static int LOOP = 100;
    
        @Override
        public void run() { 
            InParkingDataEventTranslator tradeTranslator = new InParkingDataEventTranslator();
            for(int i = 0; i < LOOP; i++){ 
                disruptor.publishEvent(tradeTranslator);
                try { 
                    Thread.sleep(1000L);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
            latch.countDown();
            System.out.println("生产者写完"+LOOP+"个消息");
        }
    
        class InParkingDataEventTranslator implements EventTranslator<InParkingDataEvent>{ 
    
            @Override
            public void translateTo(InParkingDataEvent event, long sequence) { 
                this.generateTradeTransaction(event);
            }
    
            private InParkingDataEvent generateTradeTransaction(InParkingDataEvent event){ 
                int num = (int)(Math.random()*8000);
                num = num + 1000;
                event.setCarLicense("京Z"+num);
                System.out.println("Thread Id "+Thread.currentThread().getId()+" 写完一个event");
                return event;
            }
    
        }
    
    }

package com.luke.study.disruptor.consumer;
    
    import com.lmax.disruptor.dsl.Disruptor;
    import com.lmax.disruptor.dsl.EventHandlerGroup;
    import com.luke.study.disruptor.event.InParkingDataEvent;
    import com.luke.study.disruptor.handler.ParkingDataInDbHandler;
    import com.luke.study.disruptor.handler.ParkingDataInKafkaHandler;
    import com.luke.study.disruptor.handler.ParkingDataInSmsHandler;
    
    /** * @Descrtption 事件消费者 * @Author luke * @Date 2019/8/29 **/
    public class InParkingDataEventConsumer { 
    
        /** * 监听事件 * @param disruptor */
        public void  listenEvent(Disruptor<InParkingDataEvent> disruptor){ 
            //创建消费者组
            EventHandlerGroup<InParkingDataEvent> handlerGroup =
                    disruptor.handleEventsWith(new ParkingDataInKafkaHandler(),new ParkingDataInDbHandler());
            ParkingDataInSmsHandler smsHandler = new ParkingDataInSmsHandler();
                handlerGroup.then(smsHandler);
        }
    
    }

package com.luke.study.disruptor;
    
    import com.lmax.disruptor.YieldingWaitStrategy;
    import com.lmax.disruptor.dsl.Disruptor;
    import com.lmax.disruptor.dsl.ProducerType;
    import com.luke.study.disruptor.consumer.InParkingDataEventConsumer;
    import com.luke.study.disruptor.event.InParkingDataEvent;
    import com.luke.study.disruptor.publisher.InParkingDataEventPublisher;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /** * @Descrtption MainTest 一个生产者三个订阅者（消费者） * Disruptor：其本质是一种无锁线程间通讯框架 * Disruptor 唯一需要处理访问冲突的地方，是多个生产者写入 Ring Buffer 的场景 * 参考：https://blog.csdn.net/mawming/article/details/52582776 * @Author luke * @Date 2019/8/29 **/
    public class MainTest { 
    
        public static void main(String[] args) throws InterruptedException { 
            long beginTime = System.currentTimeMillis();
    
            //构建disruptor
            int bufferSize = 1024;
            ExecutorService executor = Executors.newFixedThreadPool(4);
            Disruptor<InParkingDataEvent> disruptor = new Disruptor<>(() -> new InParkingDataEvent(),
                    bufferSize,Executors.defaultThreadFactory(),ProducerType.SINGLE,new YieldingWaitStrategy());
    
            //消费者开始监听事件
            InParkingDataEventConsumer consumer = new InParkingDataEventConsumer();
            consumer.listenEvent(disruptor);
    
            //生产者生产消息
            disruptor.start();
            CountDownLatch latch = new CountDownLatch(1);
            executor.submit(new InParkingDataEventPublisher(disruptor,latch));
            latch.await();
    
            disruptor.shutdown();
            executor.shutdown();
            System.out.println("总耗时："+(System.currentTimeMillis()-beginTime)/1000+"秒");
    
        }
    
    }

# 例子结果 #

Thread Id 16 写完一个event  
Thread Id 13 send 京Z8890 in plaza sms to kafka  
Thread Id 14 save 京Z8890 into db …  
Thread Id 15 send 京Z8890 in plaza sms to user  
Thread Id 16 写完一个event  
Thread Id 14 save 京Z4691 into db …  
Thread Id 13 send 京Z4691 in plaza sms to kafka  
Thread Id 15 send 京Z4691 in plaza sms to user  
Thread Id 16 写完一个event  
Thread Id 13 send 京Z7250 in plaza sms to kafka  
Thread Id 14 save 京Z7250 into db …  
Thread Id 15 send 京Z7250 in plaza sms to user  
Thread Id 16 写完一个event  
Thread Id 14 save 京Z2750 into db …  
Thread Id 13 send 京Z2750 in plaza sms to kafka  
Thread Id 15 send 京Z2750 in plaza sms to user  
生产者写完100个消息  
总耗时：100秒

# 参考 #

http://ifeve.com/disruptor/  
https://www.cnblogs.com/shenck/p/4002456.html  
https://blog.csdn.net/mawming/article/details/52582776

[20190830151046218.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/3074415cf11e4c7d87caa94adaaa6e7d.png