spring-boot项目整合Disruptor的初步使用
1.在项目的pom文件中配置
<dependency> <groupId>com.lmax</groupId> <artifactId>disruptor</artifactId> <version>3.4.2</version> </dependency> <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>r09</version> </dependency>
2.创建BaseQueueHelper
/** * lmax.disruptor 高效队列处理模板. 支持初始队列,即在init()前进行发布。 * * 调用init()时才真正启动线程开始处理 系统退出自动清理资源. */ public abstract class BaseQueueHelper<D, E extends ValueWrapper<D>, H extends WorkHandler<E>> { /** * 记录所有的队列,系统退出时统一清理资源 */ private static List<BaseQueueHelper> queueHelperList = new ArrayList<BaseQueueHelper>(); /** * Disruptor 对象 */ private Disruptor<E> disruptor; /** * RingBuffer */ private RingBuffer<E> ringBuffer; /** * initQueue */ private List<D> initQueue = new ArrayList<D>(); /** * 队列大小 * * @return 队列长度,必须是2的幂 */ protected abstract int getQueueSize(); /** * 事件工厂 * * @return EventFactory */ protected abstract EventFactory<E> eventFactory(); /** * 事件消费者 * * @return WorkHandler[] */ protected abstract WorkHandler[] getHandler(); /** * 初始化 */ public void init() { ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("DisruptorThreadPool").build(); disruptor = new Disruptor<E>(eventFactory(), getQueueSize(), namedThreadFactory, ProducerType.SINGLE, getStrategy()); disruptor.setDefaultExceptionHandler(new MyHandlerException()); disruptor.handleEventsWithWorkerPool(getHandler()); ringBuffer = disruptor.start(); //初始化数据发布 for (D data : initQueue) { ringBuffer.publishEvent(new EventTranslatorOneArg<E, D>() { @Override public void translateTo(E event, long sequence, D data) { event.setValue(data); } }, data); } //加入资源清理钩子 synchronized (queueHelperList) { if (queueHelperList.isEmpty()) { Runtime.getRuntime().addShutdownHook(new Thread() { @Override public void run() { for (BaseQueueHelper baseQueueHelper : queueHelperList) { baseQueueHelper.shutdown(); } } }); } queueHelperList.add(this); } } /** * 如果要改变线程执行优先级,override此策略. YieldingWaitStrategy会提高响应并在闲时占用70%以上CPU, * 慎用SleepingWaitStrategy会降低响应更减少CPU占用,用于日志等场景. * * @return WaitStrategy */ protected abstract WaitStrategy getStrategy(); /** * 插入队列消息,支持在对象init前插入队列,则在队列建立时立即发布到队列处理. */ public synchronized void publishEvent(D data) { if (ringBuffer == null) { initQueue.add(data); return; } ringBuffer.publishEvent(new EventTranslatorOneArg<E, D>() { @Override public void translateTo(E event, long sequence, D data) { event.setValue(data); } }, data); } /** * 关闭队列 */ public void shutdown() { disruptor.shutdown(); } }
3.创建MyHandlerException
public class MyHandlerException implements ExceptionHandler { private Logger logger = LoggerFactory.getLogger(MyHandlerException.class); /* * (non-Javadoc) 运行过程中发生时的异常 * * @see * com.lmax.disruptor.ExceptionHandler#handleEventException(java.lang.Throwable * , long, java.lang.Object) */ @Override public void handleEventException(Throwable ex, long sequence, Object event) { ex.printStackTrace(); logger.error("process data error sequence ==[{}] event==[{}] ,ex ==[{}]", sequence, event.toString(), ex.getMessage()); } /* * (non-Javadoc) 启动时的异常 * * @see * com.lmax.disruptor.ExceptionHandler#handleOnStartException(java.lang. * Throwable) */ @Override public void handleOnStartException(Throwable ex) { logger.error("start disruptor error ==[{}]!", ex.getMessage()); } /* * (non-Javadoc) 关闭时的异常 * * @see * com.lmax.disruptor.ExceptionHandler#handleOnShutdownException(java.lang * .Throwable) */ @Override public void handleOnShutdownException(Throwable ex) { logger.error("shutdown disruptor error ==[{}]!", ex.getMessage()); } }
4.创建ValueWrapper
public abstract class ValueWrapper<T> { private T value; public ValueWrapper() {} public ValueWrapper(T value) { this.value = value; } public T getValue() { return value; } public void setValue(T value) { this.value = value; } }
5.创建EventFactory
public class EventFactory implements com.lmax.disruptor.EventFactory<SeriesDataEvent> { @Override public SeriesDataEvent newInstance() { return new SeriesDataEvent(); } }
- 创建DisruptorConfig
@Configuration @ComponentScan(value = {"com.demo.disruptor"}) public class DisruptorConfig { /** * smsParamEventHandler1 * * @return SeriesDataEventHandler */ @Bean public SeriesDataEventHandler smsParamEventHandler1() { return new SeriesDataEventHandler(); } /** * smsParamEventHandler2 * * @return SeriesDataEventHandler */ @Bean public SeriesDataEventHandler smsParamEventHandler2() { return new SeriesDataEventHandler(); } /** * smsParamEventHandler3 * * @return SeriesDataEventHandler */ @Bean public SeriesDataEventHandler smsParamEventHandler3() { return new SeriesDataEventHandler(); } /** * smsParamEventHandler4 * * @return SeriesDataEventHandler */ @Bean public SeriesDataEventHandler smsParamEventHandler4() { return new SeriesDataEventHandler(); } /** * smsParamEventHandler5 * * @return SeriesDataEventHandler */ @Bean public SeriesDataEventHandler smsParamEventHandler5() { return new SeriesDataEventHandler(); } /** * smsParamEventHandler5 * * @return SeriesDataEventHandler */ @Bean public SeriesDataEventHandler smsParamEventHandler6() { return new SeriesDataEventHandler(); } }
7.创建SeriesData
public class SeriesData { private String deviceInfoStr; public SeriesData() { } public SeriesData(String deviceInfoStr) { this.deviceInfoStr = deviceInfoStr; } public String getDeviceInfoStr() { return deviceInfoStr; } public void setDeviceInfoStr(String deviceInfoStr) { this.deviceInfoStr = deviceInfoStr; } @Override public String toString() { return "SeriesData{" + "deviceInfoStr='" + deviceInfoStr + '\'' + '}'; } }
8.创建SeriesDataEvent
public class SeriesDataEvent extends ValueWrapper<SeriesData> { }
9.创建SeriesDataEventQueueHelper
@Component public class SeriesDataEventQueueHelper extends BaseQueueHelper<SeriesData, SeriesDataEvent, SeriesDataEventHandler> implements InitializingBean { private static final int QUEUE_SIZE = 1024; @Autowired private List<SeriesDataEventHandler> seriesDataEventHandler; @Override protected int getQueueSize() { return QUEUE_SIZE; } @Override protected com.lmax.disruptor.EventFactory eventFactory() { return new EventFactory(); } @Override protected WorkHandler[] getHandler() { int size = seriesDataEventHandler.size(); SeriesDataEventHandler[] paramEventHandlers = (SeriesDataEventHandler[]) seriesDataEventHandler.toArray(new SeriesDataEventHandler[size]); return paramEventHandlers; } @Override protected WaitStrategy getStrategy() { return new BlockingWaitStrategy(); //return new YieldingWaitStrategy(); } @Override public void afterPropertiesSet() throws Exception { this.init(); } }
10.创建SeriesDataEventHandler
public class SeriesDataEventHandler implements WorkHandler<SeriesDataEvent> { private Logger logger = LoggerFactory.getLogger(SeriesDataEventHandler.class); @Autowired private SocketService socketService; @Override public void onEvent(SeriesDataEvent event) { if (event.getValue() == null || StringUtils.isEmpty(event.getValue().getDeviceInfoStr())) { logger.warn("receiver series data is empty!"); } logger.error("hello word!"); } }
11.使用操作
@Autowired private SeriesDataEventQueueHelper seriesDataEventQueueHelper; @Test public void demo(){ seriesDataEventQueueHelper.publishEvent(new SeriesData("hello word")); }
12.后面推出Disruptor的各种操作
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
Jackson对字符串和对象进行转换操作工具类
直接公开代码 import com.fasterxml.jackson.databind.JavaType; import com.fasterxml.jackson.databind.ObjectMapper; import java.util.List; public class ObjectandObjetUtils<T> { public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); public T objetcMapperT(Object object,Class<T> valueType) throws Exception { return OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(object),valueType); } public T stringMapperT(String string,Class<T> valueType) throws Exception { return ...
- 下一篇
mqtt消息推送
简介 近年来随着 Web 前端的快速发展,浏览器新特性层出不穷,越来越多的应用可以在浏览器端或通过浏览器渲染引擎实现,Web 应用的即时通信方式 WebSocket 得到了广泛的应用。 WebSocket 是一种在单个 TCP 连接上进行全双工通讯的协议。WebSocket 通信协议于2011年被 IETF 定为标准 RFC 6455,并由 RFC 7936 补充规范。WebSocket API 也被 W3C 定为标准。 WebSocket 使得客户端和服务器之间的数据交换变得更加简单,允许服务端主动向客户端推送数据。在 WebSocket API 中,浏览器和服务器只需要完成一次握手,两者之间就直接可以创建持久性的连接,并进行双向数据传输。 —— 摘自 维基百科 WebSocket MQTT 协议第 6 章详细约定了 MQTT 在 WebSocket [RFC6455] 连接上传输需要满足的条件,协议内容EMQ君不在此累述。由于协议实现细节较为复杂,本文选取两个常用的 JavaScript MQTT 客户端进行连接测试。 两款客户端比较 Paho.mqtt.js Paho 是 Ecl...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
-
Docker使用Oracle官方镜像安装(12C,18C,19C)
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS8编译安装MySQL8.0.19
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- MySQL8.0.19开启GTID主从同步CentOS8
- CentOS7,8上快速安装Gitea,搭建Git服务器
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
推荐阅读
最新文章
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- CentOS6,CentOS7官方镜像安装Oracle11G
- SpringBoot2整合Redis,开启缓存,提高访问速度
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- SpringBoot2整合MyBatis,连接MySql数据库做增删改查操作
- Hadoop3单机部署,实现最简伪集群
- MySQL8.0.19开启GTID主从同步CentOS8
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果