disruptor笔记之五（事件消费实战）云计算

欢迎访问我的GitHub
https://github.com/zq2599/blog_demos
【disruptor笔记之五（事件消费实战）】内容：所有原创文章分类汇总及配套源码，涉及Java、Docker、Kubernetes、DevOPS等；
《disruptor笔记》系列链接

快速入门
Disruptor类分析
环形队列的基础操作(不用Disruptor类)
事件消费知识点小结
事件消费实战
常见场景
等待策略
知识点补充(终篇)

本篇概览
本篇是《disruptor笔记》的第五篇，前文《disruptor笔记之四：事件消费知识点小结》从理论上梳理分析了独立消费和共同消费，留下了三个任务，今天就来成这些任务，即编码实现以下三个场景：

100个订单，短信和邮件系统独立消费
100个订单，邮件系统的两个邮件服务器共同消费；
100个订单，短信系统独立消费，与此同时，两个邮件服务器共同消费；

源码下载

本篇实战中的完整源码可在GitHub下载到，地址和链接信息如下表所示(https://github.com/zq2599/blo...)：

名称	链接	备注
项目主页	https://github.com/zq2599/blo...	该项目在GitHub上的主页
git仓库地址(https)	https://github.com/zq2599/blo...	该项目源码的仓库地址，https协议
git仓库地址(ssh)	git@github.com:zq2599/blog_demos.git	该项目源码的仓库地址，ssh协议

这个git项目中有多个文件夹，本次实战的源码在disruptor-tutorials文件夹下，如下图红框所示：

文章图片

disruptor-tutorials是个父工程，里面有多个module，本篇实战的module是consume-mode，如下图红框所示：

文章图片

编写公共代码

为了完成任务，编码实现上面那三个场景，咱们需要先把公共代码写好；
首先是在父工程disruptor-tutorials下面新建名为consume-mode的module，其build.gradle内容如下：

plugins { id 'org.springframework.boot' }dependencies { implementation 'org.projectlombok:lombok' implementation 'org.springframework.boot:spring-boot-starter' implementation 'org.springframework.boot:spring-boot-starter-web' implementation 'com.lmax:disruptor'testImplementation('org.springframework.boot:spring-boot-starter-test') }

springboot启动类：

package com.bolingcavalry; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class ConsumeModeApplication { public static void main(String[] args) { SpringApplication.run(ConsumeModeApplication.class, args); } }

订单事件定义：

package com.bolingcavalry.service; import lombok.Data; import lombok.NoArgsConstructor; import lombok.ToString; @Data @ToString @NoArgsConstructor public class OrderEvent {private String value; }

- 订单事件的工程类，定义事件实例如何创建：

package com.bolingcavalry.service; import com.lmax.disruptor.EventFactory; public class OrderEventFactory implements EventFactory {@Override public OrderEvent newInstance() { return new OrderEvent(); } }

订单事件生产者类，定义如何将业务信息通过事件发布到环形队列：

package com.bolingcavalry.service; import com.lmax.disruptor.RingBuffer; public class OrderEventProducer { // 存储数据的环形队列 private final RingBuffer ringBuffer; public OrderEventProducer(RingBuffer ringBuffer) { this.ringBuffer = ringBuffer; }public void onData(String content) { // ringBuffer是个队列，其next方法返回的是下最后一条记录之后的位置，这是个可用位置 long sequence = ringBuffer.next(); try { // sequence位置取出的事件是空事件 OrderEvent orderEvent = ringBuffer.get(sequence); // 空事件添加业务信息 orderEvent.setValue(content); } finally { // 发布 ringBuffer.publish(sequence); } } }

消费订单事件的短信服务，实现EventHandler接口，所以是用在独立消费的场景：

package com.bolingcavalry.service; import com.lmax.disruptor.EventHandler; import lombok.extern.slf4j.Slf4j; import java.util.function.Consumer; @Slf4j public class SmsEventHandler implements EventHandler {public SmsEventHandler(Consumer consumer) { this.consumer = consumer; }// 外部可以传入Consumer实现类，每处理一条消息的时候，consumer的accept方法就会被执行一次 private Consumer consumer; @Override public void onEvent(OrderEvent event, long sequence, boolean endOfBatch) throws Exception { log.info("短信服务 sequence [{}], endOfBatch [{}], event : {}", sequence, endOfBatch, event); // 这里延时100ms，模拟消费事件的逻辑的耗时 Thread.sleep(100); // 如果外部传入了consumer，就要执行一次accept方法 if (null!=consumer) { consumer.accept(null); } } }

消费订单事件的邮件服务，实现EventHandler接口，所以是用在独立消费的场景：

package com.bolingcavalry.service; import com.lmax.disruptor.EventHandler; import lombok.extern.slf4j.Slf4j; import java.util.function.Consumer; @Slf4j public class MailEventHandler implements EventHandler {public MailEventHandler(Consumer consumer) { this.consumer = consumer; }// 外部可以传入Consumer实现类，每处理一条消息的时候，consumer的accept方法就会被执行一次 private Consumer consumer; @Override public void onEvent(OrderEvent event, long sequence, boolean endOfBatch) throws Exception { log.info("邮件服务 sequence [{}], endOfBatch [{}], event : {}", sequence, endOfBatch, event); // 这里延时100ms，模拟消费事件的逻辑的耗时 Thread.sleep(100); // 如果外部传入了consumer，就要执行一次accept方法 if (null!=consumer) { consumer.accept(null); } } }

消费订单事件的邮件服务，实现WorkHandler接口，所以是用在共同消费的场景：

package com.bolingcavalry.service; import com.lmax.disruptor.WorkHandler; import lombok.extern.slf4j.Slf4j; import java.util.function.Consumer; @Slf4j public class MailWorkHandler implements WorkHandler {public MailWorkHandler(Consumer consumer) { this.consumer = consumer; }// 外部可以传入Consumer实现类，每处理一条消息的时候，consumer的accept方法就会被执行一次 private Consumer consumer; @Override public void onEvent(OrderEvent event) throws Exception { log.info("共同消费模式的邮件服务 : {}", event); // 这里延时100ms，模拟消费事件的逻辑的耗时 Thread.sleep(100); // 如果外部传入了consumer，就要执行一次accept方法 if (null!=consumer) { consumer.accept(null); } } }

最后，将发布和消费事件的逻辑写在一个抽象类里，但是具体如何消费事件并不在此类中实现，而是留给子类，这个抽象类中有几处要注意的地方稍后会提到：

package com.bolingcavalry.service; import com.lmax.disruptor.dsl.Disruptor; import lombok.Setter; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import javax.annotation.PostConstruct; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; public abstract class ConsumeModeService { /** * 独立消费者数量 */ public static final int INDEPENDENT_CONSUMER_NUM = 2; /** * 环形缓冲区大小 */ protected int BUFFER_SIZE = 16; protected Disruptor disruptor; @Setter private OrderEventProducer producer; /** * 统计消息总数 */ protected final AtomicLong eventCount = new AtomicLong(); /** * 这是辅助测试用的， * 测试的时候，完成事件发布后，测试主线程就用这个countDownLatch开始等待， * 在消费到指定的数量(countDownLatchGate)后，消费线程执行countDownLatch的countDown方法， * 这样测试主线程就可以结束等待了 */ private CountDownLatch countDownLatch; /** * 这是辅助测试用的， * 测试的时候，完成事件发布后，测试主线程就用这个countDownLatch开始等待， * 在消费到指定的数量(countDownLatchGate)后，消费线程执行countDownLatch的countDown方法， * 这样测试主线程就可以结束等待了 */ private int countDownLatchGate; /** * 准备一个匿名类，传给disruptor的事件处理类， * 这样每次处理事件时，都会将已经处理事件的总数打印出来 */ protected Consumer eventCountPrinter = new Consumer