01、背景
工作中遇到我的项目应用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
03、Disruptor 的外围概念
先从理解 Disruptor 的外围概念开始,来理解它是如何运作的。上面介绍的概念模型,既是畛域对象,也是映射到代码实现上的外围对象。
04、Ring Buffer
如其名,环形的缓冲区。已经 RingBuffer 是 Disruptor 中的最次要的对象,但从3.0版本开始,其职责被简化为仅仅负责对通过 Disruptor 进行替换的数据(事件)进行存储和更新。在一些更高级的利用场景中,Ring Buffer 能够由用户的自定义实现来齐全代替。
05、Sequence Disruptor
通过程序递增的序号来编号治理通过其进行替换的数据(事件),对数据(事件)的处理过程总是沿着序号一一递增解决。一个 Sequence 用于跟踪标识某个特定的事件处理者( RingBuffer/Consumer )的解决进度。
尽管一个 AtomicLong 也能够用于标识进度,但定义 Sequence 来负责该问题还有另一个目标,那就是避免不同的 Sequence 之间的CPU缓存伪共享(Flase Sharing)问题。(注:这是 Disruptor 实现高性能的关键点之一,网上对于伪共享问题的介绍曾经车载斗量,在此不再赘述)。
06、Sequencer
Sequencer 是 Disruptor 的真正外围。此接口有两个实现类 SingleProducerSequencer、MultiProducerSequencer ,它们定义在生产者和消费者之间疾速、正确地传递数据的并发算法。
07、Sequence Barrier
用于放弃对RingBuffer的 main published Sequence 和Consumer依赖的其它Consumer的 Sequence 的援用。Sequence Barrier 还定义了决定 Consumer 是否还有可解决的事件的逻辑。
08、Wait Strategy
定义 Consumer 如何进行期待下一个事件的策略。(注:Disruptor 定义了多种不同的策略,针对不同的场景,提供了不一样的性能体现)
09、Event
在 Disruptor 的语义中,生产者和消费者之间进行替换的数据被称为事件(Event)。它不是一个被 Disruptor 定义的特定类型,而是由 Disruptor 的使用者定义并指定。
10、EventProcessor
EventProcessor 持有特定消费者(Consumer)的 Sequence,并提供用于调用事件处理实现的事件循环(Event Loop)。
11、EventHandler
Disruptor 定义的事件处理接口,由用户实现,用于处理事件,是 Consumer 的真正实现。
12、Producer
即生产者,只是泛指调用 Disruptor 公布事件的用户代码,Disruptor 没有定义特定接口或类型。
13、案例-demo
举荐一个 Spring Boot 基础教程及实战示例:https://github.com/javastacks/spring-boot-best-practice
通过上面8个步骤,你就能将Disruptor Get回家啦:
1、增加pom.xml依赖
<dependency> <groupId>com.lmax</groupId> <artifactId>disruptor</artifactId> <version>3.3.4</version></dependency>
2、音讯体Model
/** * 音讯体 */@Datapublic class MessageModel { private String message;}
3、结构EventFactory
public class HelloEventFactory implements EventFactory<MessageModel> { @Override public MessageModel newInstance() { return new MessageModel(); }}
4、结构EventHandler-消费者
@Slf4jpublic class HelloEventHandler implements EventHandler<MessageModel> { @Override public void onEvent(MessageModel event, long sequence, boolean endOfBatch) { try { //这里进行1000ms是为了确定生产音讯是异步的 Thread.sleep(1000); log.info("消费者解决音讯开始"); if (event != null) { log.info("消费者生产的信息是:{}",event); } } catch (Exception e) { log.info("消费者解决音讯失败"); } log.info("消费者解决音讯完结"); }}
5、结构BeanManager
/** * 获取实例化对象 */@Componentpublic class BeanManager implements ApplicationContextAware { private static ApplicationContext applicationContext = null; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = applicationContext; } public static ApplicationContext getApplicationContext() { return applicationContext; } public static Object getBean(String name) { return applicationContext.getBean(name); } public static <T> T getBean(Class<T> clazz) { return applicationContext.getBean(clazz); }}
6、结构MQManager
@Configurationpublic class MQManager { @Bean("messageModel") public RingBuffer<MessageModel> messageModelRingBuffer() { //定义用于事件处理的线程池, Disruptor通过java.util.concurrent.ExecutorSerivce提供的线程来触发consumer的事件处理 ExecutorService executor = Executors.newFixedThreadPool(2); //指定事件工厂 HelloEventFactory factory = new HelloEventFactory(); //指定ringbuffer字节大小,必须为2的N次方(能将求模运算转为位运算提高效率),否则将影响效率 int bufferSize = 1024 * 256; //单线程模式,获取额定的性能 Disruptor<MessageModel> disruptor = new Disruptor<>(factory, bufferSize, executor, ProducerType.SINGLE, new BlockingWaitStrategy()); //设置事件业务处理器---消费者 disruptor.handleEventsWith(new HelloEventHandler()); // 启动disruptor线程 disruptor.start(); //获取ringbuffer环,用于接取生产者生产的事件 RingBuffer<MessageModel> ringBuffer = disruptor.getRingBuffer(); return ringBuffer; }
7、结构Mqservice和实现类-生产者
public interface DisruptorMqService { /** * 音讯 * @param message */ void sayHelloMq(String message);}@Slf4j@Component@Servicepublic class DisruptorMqServiceImpl implements DisruptorMqService { @Autowired private RingBuffer<MessageModel> messageModelRingBuffer; @Override public void sayHelloMq(String message) { log.info("record the message: {}",message); //获取下一个Event槽的下标 long sequence = messageModelRingBuffer.next(); try { //给Event填充数据 MessageModel event = messageModelRingBuffer.get(sequence); event.setMessage(message); 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、结构测试类及办法
Spring Boot 根底就不介绍了,举荐下这个实战教程:
https://github.com/javastacks/spring-boot-best-practice
@Slf4j@RunWith(SpringRunner.class)@SpringBootTest(classes = DemoApplication.class)public class DemoApplicationTests { @Autowired private DisruptorMqService disruptorMqService; /** * 我的项目外部应用Disruptor做音讯队列 * @throws Exception */ @Test public void sayHelloMqTest() throws Exception{ disruptorMqService.sayHelloMq("音讯到了,Hello world!"); log.info("音讯队列已发送结束"); //这里进行2000ms是为了确定是解决音讯是异步的 Thread.sleep(2000); }}
测试运行后果
2020-04-05 14:31:18.543 INFO 7274 --- [ main] c.e.u.d.d.s.Impl.DisruptorMqServiceImpl : record the message: 音讯到了,Hello world!2020-04-05 14:31:18.545 INFO 7274 --- [ main] c.e.u.d.d.s.Impl.DisruptorMqServiceImpl : 往音讯队列中增加音讯:MessageModel(message=音讯到了,Hello world!)2020-04-05 14:31:18.545 INFO 7274 --- [ main] c.e.utils.demo.DemoApplicationTests : 音讯队列已发送结束2020-04-05 14:31:19.547 INFO 7274 --- [pool-1-thread-1] c.e.u.d.disrupMq.mq.HelloEventHandler : 消费者解决音讯开始2020-04-05 14:31:19.547 INFO 7274 --- [pool-1-thread-1] c.e.u.d.disrupMq.mq.HelloEventHandler : 消费者生产的信息是:MessageModel(message=音讯到了,Hello world!)2020-04-05 14:31:19.547 INFO 7274 --- [pool-1-thread-1] c.e.u.d.disrupMq.mq.HelloEventHandler : 消费者解决音讯完结
14、总结
其实 生成者 -> 消费者 模式是很常见的,通过一些音讯队列也能够轻松做到上述的成果。不同的中央在于,Disruptor 是在内存中以队列的形式去实现的,而且是无锁的。
这也是 Disruptor 为什么高效的起因。
版权申明:本文为CSDN博主「不二天次」的原创文章,遵循CC 4.0 BY-SA版权协定,转载请附上原文出处链接及本申明。原文链接:https://blog.csdn.net/buertianci/article/details/105327031
近期热文举荐:
1.1,000+ 道 Java面试题及答案整顿(2022最新版)
2.劲爆!Java 协程要来了。。。
3.Spring Boot 2.x 教程,太全了!
4.别再写满屏的爆爆爆炸类了,试试装璜器模式,这才是优雅的形式!!
5.《Java开发手册(嵩山版)》最新公布,速速下载!
感觉不错,别忘了顺手点赞+转发哦!