结合RocketMQ 源码,带你了解并发编程的三大神器
摘要:本文结合 RocketMQ 源码,分享并发编程三大神器的相关知识点。
本文分享自华为云社区《读 RocketMQ 源码,学习并发编程三大神器》,作者:勇哥java实战分享。
这篇文章,笔者结合 RocketMQ 源码,分享并发编程三大神器的相关知识点。
1 CountDownLatch 实现网络同步请求
CountDownLatch 是一个同步工具类,用来协调多个线程之间的同步,它能够使一个线程在等待另外一些线程完成各自工作之后,再继续执行。
下图是 CountDownLatch 的核心方法:
我们可以认为它内置一个计数器,构造函数初始化计数值。每当线程执行 countDown 方法,计数器的值就会减一,当计数器的值为 0 时,表示所有的任务都执行完成,然后在 CountDownLatch 上等待的线程就可以恢复执行接下来的任务。
举例,数据库有100万条数据需要处理,单线程执行比较慢,我们可以将任务分为5个批次,线程池按照每个批次执行,当5个批次整体执行完成后,打印出任务执行的时间 。
long start = System.currentTimeMillis();ExecutorService executorService = Executors.newFixedThreadPool(10);int batchSize = 5;CountDownLatch countDownLatch = new CountDownLatch(batchSize);for (int i = 0; i < batchSize; i++) {final int batchNumber = i;executorService.execute(new Runnable() {@Overridepublic void run() {try {doSomething(batchNumber);} catch (Exception e) {e.printStackTrace();} finally {countDownLatch.countDown();}}});
}
countDownLatch.await();
System.out.println("任务执行耗时:" + (System.currentTimeMillis() - start) + "毫秒");
温习完 CountDownLatch 的知识点,回到 RocketMQ 源码。
笔者在没有接触网络编程之前,一直很疑惑,网络同步请求是如何实现的?
同步请求指:客户端线程发起调用后,需要在指定的超时时间内,等到响应结果,才能完成本次调用。如果超时时间内没有得到结果,那么会抛出超时异常。
RocketMQ 的同步发送消息接口见下图:
追踪源码,真正发送请求的方法是通讯模块的同步请求方法 invokeSyncImpl 。
整体流程:
- 发送消息线程 Netty channel 对象调用 writeAndFlush 方法后 ,它的本质是通过 Netty 的读写线程将数据包发送到内核 , 这个过程本身就是异步的;
- ResponseFuture 类中内置一个 CountDownLatch 对象 ,responseFuture 对象调用 waitRepsone 方法,发送消息线程会阻塞 ;
3.客户端收到响应命令后, 执行 processResponseCommand 方法,核心逻辑是执行 ResponseFuture 的 putResponse 方法。
该方法的本质就是填充响应对象,并调用 countDownLatch 的 countDown 方法 , 这样发送消息线程就不再阻塞。
CountDownLatch 实现网络同步请求是非常实用的技巧,在很多开源中间件里,比如 Metaq ,Xmemcached 都有类似的实现。
2 ReadWriteLock 名字服务路由管理
读写锁是一把锁分为两部分:读锁和写锁,其中读锁允许多个线程同时获得,而写锁则是互斥锁。
它的规则是:读读不互斥,读写互斥,写写互斥,适用于读多写少的业务场景。
我们一般都使用 ReentrantReadWriteLock ,该类实现了 ReadWriteLock 。ReadWriteLock 接口也很简单,其内部主要提供了两个方法,分别返回读锁和写锁 。
public interface ReadWriteLock {//获取读锁Lock readLock();//获取写锁Lock writeLock();
}
读写锁的使用方式如下所示:
1.创建 ReentrantReadWriteLock 对象 , 当使用 ReadWriteLock 的时候,并不是直接使用,而是获得其内部的读锁和写锁,然后分别调用 lock / unlock 方法 ;
private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
2.读取共享数据 ;
Lock readLock = readWriteLock.readLock();
readLock.lock();
try {// TODO 查询共享数据
} finally {readLock.unlock();
}
3.写入共享数据;
Lock writeLock = readWriteLock.writeLock();
writeLock.lock();
try {// TODO 修改共享数据
} finally {writeLock.unlock();
}
RocketMQ架构上主要分为四部分,如下图所示 :
- Producer :消息发布的角色,Producer 通过 MQ 的负载均衡模块选择相应的 Broker 集群队列进行消息投递,投递的过程支持快速失败并且低延迟。
- Consumer :消息消费的角色,支持以 push 推,pull 拉两种模式对消息进行消费。
- BrokerServer :Broker主要负责消息的存储、投递和查询以及服务高可用保证。
- NameServer :名字服务是一个非常简单的 Topic 路由注册中心,其角色类似 Dubbo 中的zookeeper,支持Broker的动态注册与发现。
NameServer 是一个几乎无状态节点,可集群部署,节点之间无任何信息同步。Broker 启动之后会向所有 NameServer 定期(每 30s)发送心跳包(路由信息),NameServer 会定期扫描 Broker 存活列表,如果超过 120s 没有心跳则移除此 Broker 相关信息,代表下线。
那么 NameServer 如何保存路由信息呢?
路由信息通过几个 HashMap 来保存,当 Broker 向 Nameserver 发送心跳包(路由信息),Nameserver 需要对 HashMap 进行数据更新,但我们都知道 HashMap 并不是线程安全的,高并发场景下,容易出现 CPU 100% 问题,所以更新 HashMap 时需要加锁,RocketMQ 使用了 JDK 的读写锁 ReentrantReadWriteLock 。
1.更新路由信息,操作写锁
2.查询主题信息,操作读锁
读写锁适用于读多写少的场景,比如名字服务,配置服务等。
3 CompletableFuture 异步消息处理
RocketMQ 主从架构中,主节点与从节点之间数据同步/复制的方式有同步双写和异步复制两种模式。
异步复制是指消息在主节点落盘成功后就告诉客户端消息发送成功,无需等待消息从主节点复制到从节点,消息的复制由其他线程完成。
同步双写是指主节点将消息成功落盘后,需要等待从节点复制成功,再告诉客户端消息发送成功。
同步双写模式是阻塞的,笔者按照 RocketMQ 4.6.1 源码,整理出主节点处理一个发送消息的请求的时序图。
整体流程:
- 生产者将消息发送到 Broker , Broker 接收到消息后,发送消息处理器 SendMessageProcessor 的执行线程池 SendMessageExecutor 线程池来处理发送消息命令;
- 执行 ComitLog 的 putMessage 方法;
- ComitLog 内部先执行 appendMessage 方法;
- 然后提交一个 GroupCommitRequest 到同步复制服务 HAService ,等待 HAService 通知 GroupCommitRequest 完成;
- 返回写入结果并响应客户端 。
我们可以看到:发送消息的执行线程需要等待消息复制从节点 , 并将消息返回给生产者才能开始处理下一个消息。
RocketMQ 4.6.1 源码中,执行线程池的线程数量是 1 ,假如线程处理主从同步速度慢了,系统在这一瞬间无法处理新的发送消息请求,造成 CPU 资源无法被充分利用 , 同时系统的吞吐量也会降低。
那么优化同步双写呢 ?
从 RocketMQ 4.7 开始,RocketMQ 引入了 CompletableFuture 实现了异步消息处理 。
- 发送消息的执行线程不再等待消息复制到从节点后再处理新的请求,而是提前生成 CompletableFuture 并返回 ;
- HAService 中的线程在复制成功后,调用 CompletableFuture 的 complete 方法,通知 remoting 模块响应客户端(线程池:PutMessageExecutor ) 。
我们分析下 RocketMQ 4.9.4 核心代码:
1.Broker 接收到消息后,发送消息处理器 SendMessageProcessor 的执行线程池 SendMessageExecutor 线程池来处理发送消息命令;
2.调用 SendMessageProcessor 的 asyncProcessRequest 方法;
3.调用 Commitlog 的 aysncPutMessage 方法写入消息 ;
这段代码中,当 commitLog 执行完 appendMessage 后, 需要执行刷盘任务和同步复制两个任务。
但这两个任务并不是同步执行,而是异步的方式。
4.复制线程复制消息后,唤醒 future ;
5.组装响应命令 ,并将响应命令返回给客户端。
为了便于理解这一段消息发送处理过程的线程模型,笔者在 RocketMQ 源码中做了几处埋点,修改 Logback 的日志配置,发送一条普通的消息,观察服务端日志。
从日志中,我们可以观察到:
- 发送消息的执行线程(图中红色)在执行完创建刷盘 Future 和同步复制 future 之后,并没有等待这两个任务执行完成,而是在结束 asyncProcessRequest 方法后就可以处理发送消息请求了 ;
- 刷盘线程和复制线程执行完各自的任务后,唤醒 future,然后通过刷盘线程组装存储结果,最后通过 PutMessageExecutor 线程池(图中黄色)将响应命令返回给客户端。
笔者一直认为:异步是更细粒度的使用系统资源的一种方式,在异步消息处理的过程中,通过 CompletableFuture 这个神器,各个线程各司其职,优雅且高效的提升了 RocketMQ 的性能。
点击关注,第一时间了解华为云新鲜技术~
结合RocketMQ 源码,带你了解并发编程的三大神器相关推荐
- RocketMQ 源码分析 —— 集成 Spring Boot
点击上方"芋道源码",选择"设为星标" 做积极的人,而不是积极废人! 源码精品专栏 原创 | Java 2020 超神之路,很肝~ 中文详细注释的开源项目 RP ...
- 《RocketMQ源码分析》NameServer如何处理Broker的连接
<RocketMQ源码分析>NameServer如何处理Broker的连接 NameServer 介绍 NameServer 功能 动态路由发现和注册 服务剔除 创建NameServerC ...
- RocketMQ源码解析之broker文件清理
原创不易,转载请注明出处 文章目录 1. broker 清理文件介绍 1.1 哪些文件需要清理 1.2 RocketMQ文件清理的机制 2.源码解析 2.1 清理commitlog 2.2 Consu ...
- RocketMQ源码(八)Broker asyncSendMessage处理消息以及自动创建Topic
此前已经梳理了RocketMQ的broker接收Producer消息的入口源码RocketMQ(七)broker接收消息入口源码_代码---小白的博客-CSDN博客 在文章的最后我们到了SendMes ...
- RocketMQ源码解析之消息消费者(consume Message)
原创不易,转载请注明出处 文章目录 前言 1.消息流程介绍 2.源码解析 2.1 并发消费 2.2 顺序消费 前言 我们在<RocketMQ源码解析之消息消费者(pullMessage)> ...
- OpenCV下的灰度直方图生成及显示的源码,带详细注释
图像处理开发需求.图像处理接私活挣零花钱,请加微信/QQ 2487872782 图像处理开发资料.图像处理技术交流请加QQ群,群号 271891601 注:本文的源码来自博客 http://blog ...
- 探秘RocketMQ源码——Series1:Producer视角看事务消息
简介:探秘RocketMQ源码--Series1:Producer视角看事务消息 1. 前言 Apache RocketMQ作为广为人知的开源消息中间件,诞生于阿里巴巴,于2016年捐赠给了Apach ...
- 要的需求 ip提取网站源码带采集 要求是PHP源码
求.ip提取网站源码带采集 要求是PHP源码.必须带采集 类似 小峰IP提取网站,安小莫IP提取,迷惘IP提取,冰封IP提取 免费类型的 不要淘宝类型的 200 转载于:https://www.cnb ...
- thinkphp三级分销小程序源码_山东谷道微信小程序商城源码带后台 公众号平台三级分销系统...
山东谷道微信小程序商城源码带后台 公众号平台三级分销系统 那么微信二级分销系统与微信三级分销系统到底有什么区别和联系呢?为什么改了个数字地位就天差地别? 1.微信分销模式等级的区别 用简洁的话来说,微 ...
最新文章
- java中char占的二进制,java数据类型与二进制详细介绍
- 开发者需要了解的WebKit
- SAP几则常用的BASIS技巧整理
- [PHP] 深度解析Nginx下的PHP框架路由实现
- 实现RTP协议的H.264视频传输系统
- LwIP之动态内存池
- 线上环境 Linux 系统调用追踪
- oracle database version,Oracle Database Version History
- 分享2个第三方社会化分享按钮及分享工具
- 如何手动合成年度夜间灯光影像
- 改变自己,永不会晚!
- 一键在线生成朋友圈转发点赞截图教程
- ubuntu dkms报错
- BIP语言教程(一)
- linux蓝屏漏洞,WIN 10漏洞:访问特定本地设备导致系统崩溃
- cocos creator切换场景闪退_#Cocos Creator# 为什么音乐音效在场景切换的时候自动停止了?...
- AutoCAD 样条曲线如何结束
- java中字节数组转换为字符串
- 亚德诺半导体收购竞争对手美信;凯西与拜耳将在华共同推广宝丽亚和启尔畅 | 美通企业日报...
- 判断字符串是否为回文的三种常用编程语言实现