学习
实践
活动
专区
工具
TVP
写文章
专栏首页勇哥编程游记读 RocketMQ 源码,学习并发编程三大神器
原创

读 RocketMQ 源码,学习并发编程三大神器

笔者是 RocketMQ 的忠实粉丝,在阅读源码的过程中,学习到了很多编程技巧。

这篇文章,笔者结合 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() {
      @Override
      public void run() {
        try {
           doSomething(batchNumber);
        } catch (Exception e) {
           e.printStackTrace();
        } finally {
           countDownLatch.countDown();
        }
      }
   });
}
countDownLatch.await();
System.out.println("任务执行耗时:" + (System.currentTimeMillis() - start) + "毫秒");

温习完 CountDownLatch 的知识点,回到 RocketMQ 源码。

笔者在没有接触网络编程之前,一直很疑惑,<strong style="font-size: inherit;line-height: inherit;color: rgb(255, 104, 39);">网络同步请求是如何实现的?</strong>

同步请求指:客户端线程发起调用后,需要在指定的超时时间内,等到响应结果,才能完成本次调用如果超时时间内没有得到结果,那么会抛出超时异常。

RocketMQ 的同步发送消息接口见下图:

追踪源码,真正发送请求的方法是通讯模块的同步请求方法 invokeSyncImpl

整体流程:

  1. 发送消息线程 Netty channel 对象调用 writeAndFlush 方法后 ,它的本质是通过 Netty 的读写线程将数据包发送到内核 , 这个过程本身就是异步的;
  2. ResponseFuture 类中内置一个 CountDownLatch 对象 ,responseFuture 对象调用 waitRepsone 方法,发送消息线程会阻塞 ;
  1. 客户端收到响应命令后, 执行 processResponseCommand 方法,核心逻辑是执行 ResponseFuture 的 putResponse 方法。

<img src="https://oscimg.oschina.net/oscnet/up-3c7698bdd308393056f6aa00fc0d9ee38fb.png" style="zoom:115%;" />

该方法的本质就是填充响应对象,并调用 countDownLatch 的 countDown 方法 , 这样发送消息线程就不再阻塞。

CountDownLatch 实现网络同步请求是非常实用的技巧,在很多开源中间件里,比如 Metaq ,Xmemcached 都有类似的实现。

2 ReadWriteLock 名字服务路由管理

读写锁是一把锁分为两部分:读锁和写锁,其中读锁允许多个线程同时获得,而写锁则是互斥锁。

它的规则是:<strong style="font-size: inherit;line-height: inherit;color: rgb(255, 104, 39);">读读不互斥,读写互斥,写写互斥</strong>,适用于读多写少的业务场景。

我们一般都使用 ReentrantReadWriteLock ,该类实现了 ReadWriteLock 。ReadWriteLock 接口也很简单,其内部主要提供了两个方法,分别返回读锁和写锁 。

 public interface ReadWriteLock {
    //获取读锁
    Lock readLock();
    //获取写锁
    Lock writeLock();
}

读写锁的使用方式如下所示:

  1. 创建 ReentrantReadWriteLock 对象 , 当使用 ReadWriteLock 的时候,并不是直接使用,而是获得其内部的读锁和写锁,然后分别调用 lock / unlock 方法 ;
private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
  1. 读取共享数据 ;
Lock readLock = readWriteLock.readLock();
readLock.lock();
try {
   // TODO 查询共享数据
} finally {
   readLock.unlock();
}
  1. 写入共享数据;
Lock writeLock = readWriteLock.writeLock();
writeLock.lock();
try {
   // TODO 修改共享数据
} finally {
   writeLock.unlock();
}

RocketMQ架构上主要分为四部分,如下图所示 :

  1. Producer :消息发布的角色,Producer 通过 MQ 的负载均衡模块选择相应的 Broker 集群队列进行消息投递,投递的过程支持快速失败并且低延迟。
  2. Consumer :消息消费的角色,支持以 push 推,pull 拉两种模式对消息进行消费。
  3. BrokerServer :Broker主要负责消息的存储、投递和查询以及服务高可用保证。
  4. NameServer :名字服务是一个非常简单的 Topic 路由注册中心,其角色类似 Dubbo 中的zookeeper,支持Broker的动态注册与发现。

NameServer 是一个几乎无状态节点,可集群部署,节点之间无任何信息同步。Broker 启动之后会向所有 NameServer 定期(每 30s)发送心跳包(<strong style="font-size: inherit;line-height: inherit;color: rgb(255, 104, 39);">路由信息</strong>),NameServer 会定期扫描 Broker 存活列表,如果超过 120s 没有心跳则移除此 Broker 相关信息,代表下线。

那么 NameServer 如何保存路由信息呢?

路由信息通过几个 HashMap 来保存,当 Broker 向 Nameserver 发送心跳包(路由信息),Nameserver 需要对 HashMap 进行数据更新,但我们都知道 HashMap 并不是线程安全的,高并发场景下,容易出现 CPU 100% 问题,所以更新 HashMap 时需要加锁,RocketMQ 使用了 JDK 的读写锁 ReentrantReadWriteLock 。

  1. 更新路由信息,操作写锁
  1. 查询主题信息,操作读锁

读写锁适用于读多写少的场景,比如名字服务,配置服务等。

3 CompletableFuture 异步消息处理

RocketMQ 主从架构中,主节点与从节点之间数据同步/复制的方式有同步双写异步复制两种模式。

异步复制是指消息在主节点落盘成功后就告诉客户端消息发送成功,无需等待消息从主节点复制到从节点,消息的复制由其他线程完成。

同步双写是指主节点将消息成功落盘后,需要等待从节点复制成功,再告诉客户端消息发送成功。

同步双写模式是阻塞的,笔者按照 RocketMQ 4.6.1 源码,整理出主节点处理一个发送消息的请求的时序图。

整体流程:

  1. 生产者将消息发送到 Broker , Broker 接收到消息后,发送消息处理器 SendMessageProcessor 的执行线程池 SendMessageExecutor 线程池来处理发送消息命令;
  2. 执行 ComitLog 的 putMessage 方法;
  3. ComitLog 内部先执行 appendMessage 方法;
  4. 然后提交一个 GroupCommitRequest 到同步复制服务 HAService ,等待 HAService 通知 GroupCommitRequest 完成;
  5. 返回写入结果并响应客户端 。

我们可以看到:<strong style="font-size: inherit;line-height: inherit;color: rgb(255, 104, 39);">发送消息的执行线程需要等待消息复制从节点 , 并将消息返回给生产者才能开始处理下一个消息</strong>。

RocketMQ 4.6.1 源码中,执行线程池的线程数量是 1 ,假如线程处理主从同步速度慢了,系统在这一瞬间无法处理新的发送消息请求,造成 CPU 资源无法被充分利用 , 同时系统的吞吐量也会降低。

那么优化同步双写呢 ?

从 RocketMQ 4.7 开始,RocketMQ 引入了 CompletableFuture 实现了异步消息处理

  1. 发送消息的执行线程不再等待消息复制到从节点后再处理新的请求,而是提前生成 CompletableFuture 并返回 ;
  2. HAService 中的线程在复制成功后,调用 CompletableFuture 的 complete 方法,通知 remoting 模块响应客户端(线程池:PutMessageExecutor ) 。

我们分析下 RocketMQ 4.9.4 核心代码:

  1. Broker 接收到消息后,发送消息处理器 SendMessageProcessor 的执行线程池 SendMessageExecutor 线程池来处理发送消息命令;
  2. 调用 SendMessageProcessor 的 asyncProcessRequest 方法;
  1. 调用 Commitlog 的 aysncPutMessage 方法写入消息 ;

这段代码中,当 commitLog 执行完 appendMessage 后, 需要执行刷盘任务同步复制两个任务。

但这两个任务并不是同步执行,而是异步的方式。

  1. 复制线程复制消息后,唤醒 future ;
  1. 组装响应命令 ,并将响应命令返回给客户端。

为了便于理解这一段消息发送处理过程的线程模型,笔者在 RocketMQ 源码中做了几处埋点,修改 Logback 的日志配置,发送一条普通的消息,观察服务端日志。

从日志中,我们可以观察到:

  1. 发送消息的执行线程(图中红色)在执行完创建刷盘 Future 和同步复制 future 之后,并没有等待这两个任务执行完成,而是在结束 asyncProcessRequest 方法后就可以处理发送消息请求了 ;
  2. 刷盘线程和复制线程执行完各自的任务后,唤醒 future,然后通过刷盘线程组装存储结果,最后通过 PutMessageExecutor 线程池(图中黄色)将响应命令返回给客户端。

笔者一直认为:异步是更细粒度的使用系统资源的一种方式,在异步消息处理的过程中,通过 CompletableFuture 这个神器,各个线程各司其职,优雅且高效的提升了 RocketMQ 的性能。


如果我的文章对你有所帮助,还请帮忙点赞、在看、转发一下,你的支持会激励我输出更高质量的文章,非常感谢!

原创声明,本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

登录 后参与评论
0 条评论

相关文章

  • 并发编程系列之FutureTask源码学习笔记

    在上一章节的学习中,我们知道了Future类的基本用法,知道了Future其实就是为了监控线程任务执行的,接着本博客继续学习FutureTask。然后什么是Fu...

    SmileNicky
  • 【高并发】学习高并发我们究竟要不要读源码?别纠结,肝就完了!

    作者个人研发的在高并发场景下,提供的简单、稳定、可扩展的延迟消息队列框架,具有精准的定时任务和延迟队列处理功能。自开源半年多以来,已成功为十几家中小型企业提供了...

    冰河
  • RocketMq消息中间件及分布式事务消息实现原理

    一次RPC变成两次RPC、内容存储和择机投递;基于消息的通信模式,从关注处理到关注通知。

    35岁程序员那些事
  • Java继承知识点

    https://note.youdao.com/share/?id=40e70c92e16edf90b6c50267c9e2ba87&type=notebook...

    chenchenchen
  • 聊聊如何学习开源项目

    工作几年的程序员同学,有了一定的项目经验,对于编程也有了自己的理解,但他们偶尔也会感到困惑,不知道接下来该如何提升自己。

    勇哥java实战分享
  • 配置 Nginx 的神器

    点击上方“芋道源码”,选择“设为星标” 管她前浪,还是后浪? 能浪的浪,才是好浪! 每天 10:33 更新文章,每天掉亿点点头发... 源码精品专栏 原创 |...

    芋道源码
  • 浅谈源码阅读

    读研期间,为了掌握嵌入式开发,读了很多Linux的内核源码,摸索清楚了一个Linux系统从系统加电到控制台出现的流程。刚毕业转Java开发那会,为了掌握后端开发...

    阿杜
  • 我是如何拿到蚂蚁金服 offer 的 ?

    进入阿里一直都是我的梦想,我想这跟很多同学是相似的。我前后被拒了两次,分别是大三找实习和工作一年后。第二次尝试一面都没过,但那一次电话面试经历了一个半小时,被问...

    良月柒
  • SpringBoot + Redis:模拟 30w 人的秒杀抢单!

    在电商业务中,秒杀属于技术挑战最大的业务,只有经验够丰富、底子够稳的程序员,才能够hold住从搭建、上线到调优全链路。 双十一就是一个经典的秒杀案例,动辄数十万...

    江南一点雨
  • 关于你们最近的提问

    最近比较奇怪,收到不少在微信问我怎么学习语言,看什么书才好的,还有问我是不是要换专业、转行做IT应该怎么学习的?

    艾小仙
  • 想要快速进阶Java架构师?这份超强(长)学习计划单 请签收!

    而Java程序员一向比别人更难,如果说大家都在修仙的话,java程序员简直神似“剑修”,入行枯燥精通难,要想变得强大,需要能力也需要运气,更需要持之以恒。

    秃顶的Java程序员
  • 如何把开源项目用好?图解 RocketMQ 核心原理

    本文来自 Apache RocketMQ 的资深用户丁威,他和 MyCat 的核心开发者周继锋合著了《RocketMQ技术内幕:架构设计与实现原理》一书,目的是...

    华章科技
  • 源码阅读技巧篇

    主题 为什么要读源码 读什么样的源码 有什么技巧 思考、交流 坚持 为什么要读源码 说到读源码,让我想起来了读书,古语有云:“读破万卷书,下笔如有神”。 多读...

    程序猿DD
  • 我的另一种参与 RocketMQ 开源社区的方式

    首先先“SHOW”一波我在 RocketMQ 开源社区所获得的成就:2019年RocketMQ社区授予我优秀布道师荣誉称号,证书很高大上,奖品丰厚哦。

    丁威
  • 又一里程碑!阿里首推Java技术成长笔记,业内评级“钻石级”

    根据数据表明,阿里巴巴已经连续3年获评最受欢迎的中国互联网公司,实际上阿里巴巴无论在科技创新力还是社会创造价值这几个方面,都是具有一定代表里的。在行业内,很多互...

    愿天堂没有BUG

扫码关注腾讯云开发者

领取腾讯云代金券