工作中遇到项目使用Disruptor做消息队列,对你没看错,不是Kafka,也不是rabbitmq;Disruptor有个最大的优点就是快,还有一点它是开源的哦,下面做个简单的记录。
【资料图】
02、Disruptor介绍
Disruptor 是英国外汇交易公司LMAX开发的一个高性能队列,研发的初衷是解决内存队列的延迟问题(在性能测试中发现竟然与I/O操作处于同样的数量级)。基于 Disruptor 开发的系统单线程能支撑每秒 600 万订单,2010 年在 QCon 演讲后,获得了业界关注。
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 的核心概念开始,来了解它是如何运作的。下面介绍的概念模型,既是领域对象,也是映射到代码实现上的核心对象。
如其名,环形的缓冲区。曾经 RingBuffer 是 Disruptor 中的最主要的对象,但从3.0版本开始,其职责被简化为仅仅负责对通过 Disruptor 进行交换的数据(事件)进行存储和更新。在一些更高级的应用场景中,Ring Buffer 可以由用户的自定义实现来完全替代。
通过顺序递增的序号来编号管理通过其进行交换的数据(事件),对数据(事件)的处理过程总是沿着序号逐个递增处理。一个 Sequence 用于跟踪标识某个特定的事件处理者( RingBuffer/Consumer )的处理进度。虽然一个 AtomicLong 也可以用于标识进度,但定义 Sequence 来负责该问题还有另一个目的,那就是防止不同的 Sequence 之间的CPU缓存伪共享(Flase Sharing)问题。(注:这是 Disruptor 实现高性能的关键点之一,网上关于伪共享问题的介绍已经汗牛充栋,在此不再赘述)。
Sequencer 是 Disruptor 的真正核心。此接口有两个实现类 SingleProducerSequencer、MultiProducerSequencer ,它们定义在生产者和消费者之间快速、正确地传递数据的并发算法。
用于保持对RingBuffer的 main published Sequence 和Consumer依赖的其它Consumer的 Sequence 的引用。Sequence Barrier 还定义了决定 Consumer 是否还有可处理的事件的逻辑。
定义 Consumer 如何进行等待下一个事件的策略。(注:Disruptor 定义了多种不同的策略,针对不同的场景,提供了不一样的性能表现)
在 Disruptor 的语义中,生产者和消费者之间进行交换的数据被称为事件(Event)。它不是一个被 Disruptor 定义的特定类型,而是由 Disruptor 的使用者定义并指定。
EventProcessor 持有特定消费者(Consumer)的 Sequence,并提供用于调用事件处理实现的事件循环(Event Loop)。
Disruptor 定义的事件处理接口,由用户实现,用于处理事件,是 Consumer 的真正实现。
即生产者,只是泛指调用 Disruptor 发布事件的用户代码,Disruptor 没有定义特定接口或类型。
com.lmax disruptor 3.3.4
2、消息体Model /***消息体*/@DatapublicclassMessageModel{privateStringmessage;}
3、构造EventFactory publicclassHelloEventFactoryimplementsEventFactory{@OverridepublicMessageModelnewInstance(){returnnewMessageModel();}}
4、构造EventHandler-消费者 @Slf4jpublicclassHelloEventHandlerimplementsEventHandler{@OverridepublicvoidonEvent(MessageModelevent,longsequence,booleanendOfBatch){try{//这里停止1000ms是为了确定消费消息是异步的Thread.sleep(1000);log.info(\"消费者处理消息开始\");if(event!=null){log.info(\"消费者消费的信息是:{}\",event);}}catch(Exceptione){log.info(\"消费者处理消息失败\");}log.info(\"消费者处理消息结束\");}}
5、构造BeanManager /***获取实例化对象*/@ComponentpublicclassBeanManagerimplementsApplicationContextAware{privatestaticApplicationContextapplicationContext=null;@OverridepublicvoidsetApplicationContext(ApplicationContextapplicationContext)throwsBeansException{this.applicationContext=applicationContext;}publicstaticApplicationContextgetApplicationContext(){returnapplicationContext;}publicstaticObjectgetBean(Stringname){returnapplicationContext.getBean(name);}publicstaticTgetBean(Classclazz){returnapplicationContext.getBean(clazz);}}
6、构造MQManager @ConfigurationpublicclassMQManager{@Bean(\"messageModel\")publicRingBuffermessageModelRingBuffer(){//定义用于事件处理的线程池,Disruptor通过java.util.concurrent.ExecutorSerivce提供的线程来触发consumer的事件处理ExecutorServiceexecutor=Executors.newFixedThreadPool(2);//指定事件工厂HelloEventFactoryfactory=newHelloEventFactory();//指定ringbuffer字节大小,必须为2的N次方(能将求模运算转为位运算提高效率),否则将影响效率intbufferSize=1024*256;//单线程模式,获取额外的性能Disruptordisruptor=newDisruptor<>(factory,bufferSize,executor,ProducerType.SINGLE,newBlockingWaitStrategy());//设置事件业务处理器---消费者disruptor.handleEventsWith(newHelloEventHandler());//启动disruptor线程disruptor.start();//获取ringbuffer环,用于接取生产者生产的事件RingBufferringBuffer=disruptor.getRingBuffer();returnringBuffer;}
7、构造Mqservice和实现类-生产者 publicinterfaceDisruptorMqService{/***消息*@parammessage*/voidsayHelloMq(Stringmessage);}@Slf4j@Component@ServicepublicclassDisruptorMqServiceImplimplementsDisruptorMqService{@AutowiredprivateRingBuffermessageModelRingBuffer;@OverridepublicvoidsayHelloMq(Stringmessage){log.info(\"recordthemessage:{}\",message);//获取下一个Event槽的下标longsequence=messageModelRingBuffer.next();try{//给Event填充数据MessageModelevent=messageModelRingBuffer.get(sequence);event.setMessage(message);log.info(\"往消息队列中添加消息:{}\",event);}catch(Exceptione){log.error(\"failedtoaddeventtomessageModelRingBufferfor:e={},{}\",e,e.getMessage());}finally{//发布Event,激活观察者去消费,将sequence传递给改消费者//注意最后的publish方法必须放在finally中以确保必须得到调用;如果某个请求的sequence未被提交将会堵塞后续的发布操作或者其他的producermessageModelRingBuffer.publish(sequence);}}}
8、构造测试类及方法 @Slf4j@RunWith(SpringRunner.class)@SpringBootTest(classes=DemoApplication.class)publicclassDemoApplicationTests{@AutowiredprivateDisruptorMqServicedisruptorMqService;/***项目内部使用Disruptor做消息队列*@throwsException*/@TestpublicvoidsayHelloMqTest()throwsException{disruptorMqService.sayHelloMq(\"消息到了,Helloworld!\");log.info(\"消息队列已发送完毕\");//这里停止2000ms是为了确定是处理消息是异步的Thread.sleep(2000);}}
测试运行结果
2020-04-0514:31:18.543INFO7274---[main]c.e.u.d.d.s.Impl.DisruptorMqServiceImpl:recordthemessage:消息到了,Helloworld!2020-04-0514:31:18.545INFO7274---[ main] c.e.u.d.d.s.Impl.DisruptorMqServiceImpl :往消息队列中添加消息:MessageModel(message=消息到了,Hello world!)2020-04-0514:31:18.545INFO7274---[main]c.e.utils.demo.DemoApplicationTests:消息队列已发送完毕2020-04-0514:31:19.547INFO7274---[pool-1-thread-1]c.e.u.d.disrupMq.mq.HelloEventHandler:消费者处理消息开始2020-04-0514:31:19.547INFO7274---[pool-1-thread-1] c.e.u.d.disrupMq.mq.HelloEventHandler :消费者消费的信息是:MessageModel(message=消息到了,Hello world!)2020-04-0514:31:19.547INFO7274---[pool-1-thread-1]c.e.u.d.disrupMq.mq.HelloEventHandler:消费者处理消息结束
其实 生成者 -> 消费者 模式是很常见的,通过一些消息队列也可以轻松做到上述的效果。不同的地方在于,Disruptor 是在内存中以队列的方式去实现的,而且是无锁的。这也是 Disruptor 为什么高效的原因。
全球观天下!Spring Boot + Disruptor = 王炸!!
寻女30载,元宵节前她回家了 全球聚看点
当前热文:防守牛,起伏大!李怡婧/罗徐敏1比2负于李汶妹/刘玄炫,无缘4强
环球精选!这宝马的敞篷跑车,全进口+后驱,配置媲美百万级座驾
足协官方:2023年中国足协国际级裁判员名单
阳光凸显雨水“贵” 2023年1月四川降水量为历史同期最少
【世界聚看点】跟着摇滚乐队“痛仰之歌”,一起唱响张掖
快资讯:排放超标吗?现场测测 新乡市建成机动车“天地人车”一体化监控系统
朗明纳斯:光+未来,用光子改变世界_全球速读
北京城市副中心生机勃发,“千里潞道”串起蓝绿空间
解读哈弗H6S潮跑版4WD,售15.99万,造型动感,配置实用 世界动态
组织管理_关于组织管理介绍 每日观察
一晃孩子长大了的句子短句英语(优选105句) 头条
世界热资讯!微信账号解冻客服_微信账号解冻客服电话
谷歌今天让购买Pixel6成为不可能
世界讯息:开年上演现象级“出圈” 《狂飙》“狂飙”的背后
今日热门!美国重新开放驻所罗门群岛的大使馆,外交部回应
2023兰州普通话水平测试报名时间+报名入口 天天播报
2月2日华南地区苯酐市场走势暂稳
【少前同人】一名来自中国的指挥官的逃亡日记(35+36)【物理补档】|环球速读
卡梅隆·史密斯、达斯汀·约翰逊、德尚博强势开启亚巡新赛季,中国小将丁文一首次出战!明日16:00,高尔夫频道直播!|全球观速讯
qq飞车手游_qq医生
天天消息!像特纳也像大洛佩兹!湖人签了一个什么首发?受益的不止詹姆斯
新春走基层▏放不下小区的独居老人们,物业经理连续7年在岗位上过年
【环球时快讯】元一传媒(835452):拟续聘会计师事务所公告
北交所券商执业格局初现:头部逐步加码 20家券商获评一档|环球讯息
海南省东方市乐妹村—— 缕缕花香富农家(新春走基层) 热文
莱绅通灵(603900)2月1日9点37分触及涨停板
世界快看点丨中性洗涤剂有哪些洗杯子_中性洗涤剂有哪些
ABC是哪个国家的品牌_ABC是什么国家的品牌