Disruptor系列3：Disruptor样例实战

2024-04-29 09:18
文章标签 实战 系列 样例 disruptor

本文主要是介绍Disruptor系列3：Disruptor样例实战，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

章节回顾：
- Disruptor系列1：初识Disruptor
- Disruptor系列2：Disruptor原理剖析

本章节是Disruptor样例实战，依据Disruptor的工作流依次执行的特性，实现各种样例。如果想了解Disruptor是什么，可以查看章节 Disruptor系列1：初识Disruptor ，如果想深层次了解Disruptor，可以查看章节 Disruptor系列2：Disruptor原理剖析。通过本章节，希望让大家对如何使用Disruptor有个初步认识，看看它能够解决哪些情况。

具体而言，它可以解决如下方面：
- 并行计算实现；
- 串行依次执行；
- 菱形方式执行；
- 链式并行计算。
并且基于以上情况，每种类型的消费者都可以池化，默认初始化多个同一类型的消费者实例，并行处理，提高系统吞吐量。

本样例是一个生产者生产一个Long类型的数值，消费者对该数值进行处理的操作。本样例对以上各种情况的实现只是disruptor注册消费者的方式不同，因此，我们先把事件类、事件工厂类、消费者类、事件转换类和主函数贴出来。

事件类

public class LongEvent {private Long number;public Long getNumber() {return number;}public void setNumber(Long number) {this.number = number;}
}

事件工厂类

public class LongEventFactory implements EventFactory<LongEvent> {@Overridepublic LongEvent newInstance() {return new LongEvent();}
}

事件转换类

public class LongEventTranslator implements EventTranslatorOneArg<LongEvent, Long> {@Overridepublic void translateTo(LongEvent event, long sequence, Long arg0) {event.setNumber(arg0);}
}

C1-1消费者类

该消费者执行将数值+10的操作。可以看到该消费者同时实现了EventHandler和WorkHandler两个接口。如果不需要池化，只需要实现EventHandler类即可。如果需要池化，只需要实现WorkHandler类即可。本例为了能够同时讲解池化和非池化的实现，因此同时实现了两个类，当然，也没啥问题。

public class C11EventHandler implements EventHandler<LongEvent>,WorkHandler<LongEvent> {@Overridepublic void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {long number = event.getNumber();number += 10;System.out.println(System.currentTimeMillis()+": c1-1 consumer finished.number=" + number);}@Overridepublic void onEvent(LongEvent event) throws Exception {long number = event.getNumber();number += 10;System.out.println(System.currentTimeMillis()+": c1-1 consumer finished.number=" + number);}
}

C1-2消费者类

该消费者类执行将数值乘以10的操作。

public class C12EventHandler implements EventHandler<LongEvent>,WorkHandler<LongEvent> {@Overridepublic void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {long number = event.getNumber();number *= 10;System.out.println(System.currentTimeMillis()+": c1-2 consumer finished.number=" + number);}@Overridepublic void onEvent(LongEvent event) throws Exception {long number = event.getNumber();number *= 10;System.out.println(System.currentTimeMillis()+": c1-2 consumer finished.number=" + number);}
}

c2-1消费者类

该消费者类负责将数值+20.

public class C21EventHandler implements EventHandler<LongEvent>,WorkHandler<LongEvent> {@Overridepublic void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {long number = event.getNumber();number += 20;System.out.println(System.currentTimeMillis()+": c2-1 consumer finished.number=" + number);}@Overridepublic void onEvent(LongEvent event) throws Exception {long number = event.getNumber();number += 20;System.out.println(System.currentTimeMillis()+": c2-1 consumer finished.number=" + number);}
}

C2-2消费者类

该消费者类负责将数值*20

public class C22EventHandler implements EventHandler<LongEvent>,WorkHandler<LongEvent> {@Overridepublic void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {long number = event.getNumber();number *= 20;System.out.println(System.currentTimeMillis()+": c2-2 consumer finished.number=" + number);}@Overridepublic void onEvent(LongEvent event) throws Exception {long number = event.getNumber();number *= 20;System.out.println(System.currentTimeMillis()+": c2-2 consumer finished.number=" + number);}
}

主函数

public class Main {public static void main(String[] args) {int bufferSize = 1024*1024;//环形队列长度，必须是2的N次方EventFactory<LongEvent> eventFactory = new LongEventFactory();/*** 定义Disruptor，基于单生产者，阻塞策略*/Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(eventFactory,bufferSize, Executors.defaultThreadFactory(),ProducerType.SINGLE,new BlockingWaitStrategy());/////////////////////////////////////////////////////////////////////XXX(disruptor);//这里是调用各种不同方法的地方./////////////////////////////////////////////////////////////////////RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();/*** 输入10*/ringBuffer.publishEvent(new LongEventTranslator(),10L);ringBuffer.publishEvent(new LongEventTranslator(),100L);}
}

并行计算实现

这里写图片描述
并行计算就是消费者之间互相不依赖，并行执行，执行开始时间是一样的。

/*** 并行计算实现,c1,c2互相不依赖* <br/>* p --> c11*   --> c21*/public static void parallel(Disruptor<LongEvent> disruptor){disruptor.handleEventsWith(new C11EventHandler(),new C21EventHandler());disruptor.start();}

串行计算，依次执行

这里写图片描述

/*** 串行依次执行* <br/>* p --> c11 --> c21* @param disruptor*/public static void serial(Disruptor<LongEvent> disruptor){disruptor.handleEventsWith(new C11EventHandler()).then(new C21EventHandler());disruptor.start();}

菱形方式执行

这里写图片描述

/*** 菱形方式执行* <br/>*   --> c11* p          --> c21*   --> c12* @param disruptor*/public static void diamond(Disruptor<LongEvent> disruptor){disruptor.handleEventsWith(new C11EventHandler(),new C12EventHandler()).then(new C21EventHandler());disruptor.start();}

链式并行计算

这里写图片描述

/*** 链式并行计算* <br/>*   --> c11 --> c12* p*   --> c21 --> c22* @param disruptor*/public static void chain(Disruptor<LongEvent> disruptor){disruptor.handleEventsWith(new C11EventHandler()).then(new C12EventHandler());disruptor.handleEventsWith(new C21EventHandler()).then(new C22EventHandler());disruptor.start();}

上面的实例，每一种消费者都只有一个实例，如果想多个实例形成一个线程池并发处理多个任务怎么办？如果使用disruptor.handleEventWith(new C11EventHandler(),new C11EventHandler(),...)这种，会造成重复消费同一个数据，不是我们想要的。我们想要的是同一个类的实例消费不同的数据，怎么办？
- 首先，消费者类需要实现WorkHandler接口，而不是EventHandler接口。为了方便，我们同时实现了这两个接口。
- 其次，disruptor调用handleEventsWithWorkerPool方法，而不是handleEventsWith方法
- 最后，实例化多个事件消费类。

并行计算实现,c1,c2互相不依赖,同时C1，C2分别有2个实例

这里写图片描述

/*** 并行计算实现,c1,c2互相不依赖,同时C1，C2分别有2个实例* <br/>* p --> c11*   --> c21*/public static void parallelWithPool(Disruptor<LongEvent> disruptor){disruptor.handleEventsWithWorkerPool(new C11EventHandler(),new C11EventHandler());disruptor.handleEventsWithWorkerPool(new C21EventHandler(),new C21EventHandler());disruptor.start();}

串行依次执行,同时C11，C21分别有2个实例

这里写图片描述

/*** 串行依次执行,同时C11，C21分别有2个实例* <br/>* p --> c11 --> c21* @param disruptor*/public static void serialWithPool(Disruptor<LongEvent> disruptor){disruptor.handleEventsWithWorkerPool(new C11EventHandler(),new C11EventHandler()).then(new C21EventHandler(),new C21EventHandler());disruptor.start();}

这篇关于Disruptor系列3：Disruptor样例实战的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/945690。

相关文章

React+TS前台项目实战（十七）-- 全局常用组件Dropdown封装

React+TS前台项目实战（十七）-- 全局常用组件Dropdown封装

文章目录 前言Dropdown组件1. 功能分析2. 代码+详细注释3. 使用方式4. 效果展示 总结 前言 今天这篇主要讲全局Dropdown组件封装，可根据UI设计师要求自定义修改。 Dropdown组件 1. 功能分析 （1）通过position属性,可以控制下拉选项的位置 （2）通过传入width属性, 可以自定义下拉选项的宽度 （3）通过传入classN
阅读更多...
一道经典Python程序样例带你飞速掌握Python的字典和列表

一道经典Python程序样例带你飞速掌握Python的字典和列表

Python中的列表（list）和字典（dict）是两种常用的数据结构，它们在数据组织和存储方面有很大的不同。 列表（List） 列表是Python中的一种有序集合，可以随时添加和删除其中的元素。列表中的元素可以是任何数据类型，包括数字、字符串、其他列表等。列表使用方括号[]表示，元素之间用逗号,分隔。 定义和使用 # 定义一个列表 fruits = ['apple', 'banana
阅读更多...
JavaWeb系列二十: jQuery的DOM操作 下

JavaWeb系列二十: jQuery的DOM操作 下

jQuery的DOM操作 CSS-DOM操作多选框案例页面加载完毕触发方法作业布置jQuery获取选中复选框的值jQuery控制checkbox被选中jQuery控制(全选/全不选/反选)jQuery动态添加删除用户 CSS-DOM操作 获取和设置元素的样式属性: css()获取和设置元素透明度: opacity属性获取和设置元素高度, 宽度: height(), widt
阅读更多...
C语言入门系列：探秘二级指针与多级指针的奇妙世界

C语言入门系列：探秘二级指针与多级指针的奇妙世界

文章目录 一，指针的回忆杀1，指针的概念2，指针的声明和赋值3，指针的使用3.1 直接给指针变量赋值3.2 通过*运算符读写指针指向的内存3.2.1 读3.2.2 写 二，二级指针详解1，定义2，示例说明3，二级指针与一级指针、普通变量的关系3.1，与一级指针的关系3.2，与普通变量的关系,示例说明 4，二级指针的常见用途5，二级指针扩展到多级指针 小结 C语言的学习之旅中，二级
阅读更多...
PyTorch模型_trace实战：深入理解与应用

PyTorch模型_trace实战：深入理解与应用

pytorch使用trace模型 1、使用trace生成torchscript模型2、使用trace的模型预测 1、使用trace生成torchscript模型 def save_trace(model, input, save_path):traced_script_model = torch.jit.trace(model, input)<
阅读更多...
JavaWeb系列六: 动态WEB开发核心(Servlet) 上

JavaWeb系列六: 动态WEB开发核心(Servlet) 上

韩老师学生 官网文档为什么会出现Servlet什么是ServletServlet在JavaWeb项目位置Servlet基本使用Servlet开发方式说明快速入门- 手动开发 servlet浏览器请求Servlet UML分析Servlet生命周期GET和POST请求分发处理通过继承HttpServlet开发ServletIDEA配置ServletServlet注意事项和细节 Servlet注
阅读更多...
C语言入门系列：初识函数

C语言入门系列：初识函数

文章目录 一，C语言函数与数学函数的区别1，回忆杀-初中数学2，C语言中的函数 二， 函数的声明1，函数头1.1，函数名称1.2，返回值类型1.3，参数列表 2，函数体2.1，函数体2.2，return语句 三，main函数四，函数的参数与传递方式1，实参和形参1.1，函数定义（含形参）1.2，函数调用（使用实参） 2，参数传递方式2.1，值传递2.2，引用传递 五，函数原型与预声明1，
阅读更多...
MyBatis-Plus常用注解详解与实战应用

MyBatis-Plus常用注解详解与实战应用

MyBatis-Plus 是一个 MyBatis 的增强工具，在 MyBatis 的基础上只做增强不做改变，为简化开发、提高效率而生。它提供了大量的常用注解，使得开发者能够更方便地进行数据库操作。 MyBatis-Plus 提供的注解可以帮我们解决一些数据库与实体之间相互映射的问题。 @TableName @TableName 用来指定表名 在使用 MyBatis-Plus 实现基本的 C
阅读更多...
［大师C语言(第三十六篇)］C语言信号处理：深入解析与实战

［大师C语言(第三十六篇)］C语言信号处理：深入解析与实战

引言 在计算机科学中，信号是一种软件中断，它允许进程之间或进程与内核之间进行通信。信号处理是操作系统中的一个重要概念，它允许程序对各种事件做出响应，例如用户中断、硬件异常和系统调用。C语言作为一门接近硬件的编程语言，提供了强大的信号处理能力。本文将深入探讨C语言信号处理的技术和方法，帮助读者掌握C语言处理信号的高级技巧。 第一部分：C语言信号处理基础 1.1 信号的概念 在Unix-lik
阅读更多...
django学习入门系列之第三点《案例 小米商城头标》

django学习入门系列之第三点《案例 小米商城头标》

文章目录 阴影案例 小米商城头标往期回顾 阴影 设置阴影 box-shadow:水平方向 垂直方向 模糊距离 颜色 box-shadow: 5px 5px 5px #aaa; 案例 小米商城头标 目标样式： CSS中的代码 /*使外边距等于0，即让边框与界面贴合*/body{margin: 0;}/*控制父级边框*/.header{backgroun
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2