博客

消息中间件-RocketMQ（基础、实战、源码、原理看这一篇就够了）

  • Post author:
  • Post category:其他


基础篇

消息中间件概述

消息中间件和消息队列是同一个概念的两种名称，是分布式系统中的一个重要组件。

举个例子：比如现在有一个商城系统使用的是微服务架构，其中有订单服务、库存服务、物流服务、积分服务。业务场景是如果下单了一个商品需要调用库存服务减库存、需要调用物流服务发送商品、需要调用积分服务添加响应的积分。这中间牵扯到了服务和服务之间的调用问题，为了解决这两种问题有两种方案：

  • 通过RPC框架（也就是http请求）

  • 通过消息中间件

    消息中间件就是生产者只需要将需要处理的消息丢进来即可不需要等到消息执行完成的回调，然后其他服务再去监听消息，有消息来了就执行消息

消息中间件的应用场景

  • 异步解耦

    比如上文说到的例子，一个请求如果使用RPC框架同步处理的话是需要400ms的，如果是使用消息中间件异步处理这些消息的话只需要110ms。

    还有一种情况是，加入此时积分服务挂掉了，那么这个订单的请求就不能完成，可能造成订单下了但是积分没有加上去的bug、要么就是订单下不成功,加入使用的是消息中间件,即使积分服务挂了，但是消息成功加入到了消息队列，这次加积分的消息就一直没有被消费掉，等到积分服务重启的时候会再去消息队列中读取消息并处理对应的积分业务。

  • 削峰填谷

    如上图所示用户的请求是根据时间段的不同有所不同的，但是服务器的处理事务的能力限于配置的问题肯定也是有限的比如上图所示的1w/qps，如果用户的请求处于1w/qps一下还好说服务器都能处理的过来，如果超过1w/qps服务器就没办法处理，此时处理不掉的请求可能就会直接丢掉，但是如果引入消息中间件，用户发过来的请求不要直接给系统让系统处理，而是添加到消息中间件，那么如果用户请求超过1w/qps之后多余的消息就会丢到MQ中，等待被处理，当用户请求降下来了此时系统再赶紧处理MQ中还没处理的消息，这样处理的结果是用户端可能慢一点收到响应但是不至于报错。所以此时消息中间件就提到了削峰填谷的作用

  • 消息分发

    如上图所示：当前有个系统有个商家端还有三个用户端，商家端发布一个商品需要在三个用户端上显示，如果此时商家修改一个商品的价格，三个系统都需要同步修改对应的价格。此时就可以使用消息中间件，商家端修改了价格只需要往MQ中丢三条消息（包含商品的最终价格），然后三个分系统分别监听，拿到消息时候在修改对应的价格（应为消息中就有对应的价格所以这里还不需要访问数据库）

常见的一些消息中间件

  • ActiveMQ
  • RabbitMQ
  • KafKa
  • RocketMQ

消息中间件对比

RocketMQ的核心组件

  • 运行模型

  • Producer

    负责生产消息，生产完消息会丢到MQ中
  • Comsumer

    负责消费消息，从MQ中读取对应的消息并执行相应的任务
  • Message

    MQ中的最小单位，也就是Producer需要传输的内容载体
  • Queue

    生产者发送的消息会放到队列中，然后消费者读取队列中的消息并执行对应任务
  • Topic 表示一类消息的集合，每个主题包含若干个队列，队列中又包含着若干个消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位。
  • Broker

    就是一个代理服务器,消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。
  • NameServer

    名称服务充当路由消息的提供者。生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个NameServer实例组成集群，但相互独立，没有信息交换。

单机环境安装

这里采取的是二进制文件安装（安装的是4.7.0版本）

  1. 下载二进制文件 wget

    archive.apache.org/dist/rocket…
  2. 解压 unzip rocketmq-all-4.7.0-bin-release.zip
  3. 移动文件夹 mv rocketmq-all-4.7.0-bin-release rocketmq-4.7
  4. 修改配置参数

    为了保证RocketMQ可以正常启动, 默认情况会使用比较大的内存, 建议给NameServer和Broker设置1G的内存。修改runbroker.sh和runserver.sh脚本如下图：

  5. 启动NameServer:nohup bin/mqnamesrv &
  6. 启动Broker:nohup bin/mqbroker -n 127.0.0.1:9876 &
  7. 查看状态 jdk命令 jps
  8. 关闭服务 bin/mqshutdown broker 、bin/mqshutdown nameserver

管理控制台的安装：

  1. 下载项目 git clone

    gitee.com/heshengjun/…
  2. 进入到管理控制台项目 cd rocketmq-externals/rocketmq-console
  3. 编译项目:mvn package -Dmaven.test.skip=true
  4. 启动目录创建配置文件application.properties（配置端口和nameserver地址）

    server.port=9999

    rocketmq.config.namesrvAddr=127.0.0.1:9876
  5. 启动控制台：nohup java -jar rocketmq-console-ng-1.0.1.jar &

基本使用

  • 入门案例（同步发送消息）

    • 生产者

      1. 创建生产者（

        DefaultMQProducer

        ）
      2. 指定NameServer的地址
      3. 启动生产者
      4. 创建消息对象（

        Message

        ）
      5. 发送消息
      6. 关闭应用程序 
      //1 创建一个生产者
DefaultMQProducer producer = new DefaultMQProducer("tudou");
//2 指定NameServer的地址
producer.setNamesrvAddr("192.168.88.130:9876");
//3 启动生产者
producer.start();
for (int i = 0; i < 100; i++) {
    // 4 创建消息对象
    Message msg = new Message("tudou", ("hello tudou" + i).getBytes());
    // 5 发送消息
    SendResult result = producer.send(msg);
    System.out.println(result.getSendStatus());
    System.out.println("result.getMsgId() = " + result.getMsgId());
}
//6 关闭应用程序
producer.shutdown();
复制代码

      控制台结果：

    • 消费者

      1. 创建一个消费者对象
      2. 设置NameServer的地址
      3. 指定消费的主题
      4. 指定从哪个位置开始消费
      5. 指定一个监听器, 并发消费消息
      6. 启动消费者 
     //1. 创建消费者对象
 DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("tudou");
 //2. 设置NameServer地址
 defaultMQPushConsumer.setNamesrvAddr("192.168.88.130:9876");
 //3. 指定消费主题
 defaultMQPushConsumer.subscribe("tudou","*");
 //4. 冲那个位置开始消费
 defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
 //5. 指定一个监听器，并发消费消息
 defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
         for (MessageExt msg:list) {
             System.out.println("msgId =" + msg.getMsgId() + "----- msgTopic =" +  msg.getTopic() + "----- msgBody =" +  new String(msg.getBody()));
         }
         //这里需要返回消费成功告诉消息中间件否则该条消息还会推送给你
         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
     }
 });
 //6. 启动消费者
 defaultMQPushConsumer.start();
复制代码

    控制台效果：

  • 发送消息

    下面就是发送消息的两种方式

    • 同步消息

      这里的同步消息并不是说等到消息被消费后的同步，而是消息丢到消息中间件后的结果同步。只有等到上一个消息返回状态之后才能继续发送下一条消息。 上文中说到的例子就是一个同步发送消息的案例

      如下图所示：

      如上图所示send方法会返回一个

      SendResult

      对象则是同步消息

    • 异步消息

      发送消息不需要等到上一条消息发送完成，但是也会给到一个发送状态的回调 

        producer.send(msg,new SendCallback(){

      @Override
      public void onSuccess(SendResult sendResult) {
          //消息发送成功
      }

      @Override
      public void onException(Throwable throwable) {
          //发送消息异常
      }
  });
复制代码

    • 一次性发送消息

      生产者只需要将消息丢出去就行，不会受到发送消息后的状态，一般用于日志发送等，有点像UDP，只管把你跑出去剩下的我不管


      producer.sendOneway(msg);

    • 生产者组

      同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。如果发送的是事务消息且原始生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消费。

    • 消息封装（Tag、Key）

      Tag：用来给消息进行标记, 可以通过Tag对消息进行分类,

      Key: 可以设置消息的一个唯一ID, 用于区分每个消息的标志, 业务ID 并且在管理控制台 , 可以通过Key进行消息的查询跟踪把不同类型的消息交给不同的消费者进行消费

  • 接收消息

    • 消费方式

      • 拉式消费

        就是每次消费者主动的拉取消息队列中的消息进行消费

        DefaultLitePullConsumer defaultLitePullConsumer = new DefaultLitePullConsumer("tudou1");
defaultLitePullConsumer.setNamesrvAddr("192.168.88.130:9876");
//3. 指定消费主题
defaultLitePullConsumer.subscribe("tudou","*");
//4. 冲那个位置开始消费
defaultLitePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
defaultLitePullConsumer.start();
while (true){
    List<MessageExt> msgs = defaultLitePullConsumer.poll();
    for (MessageExt msg : msgs) {
        System.out.println("接收到的消息:" + new String(msg.getBody()));
    }
}
复制代码
      • 推式消费（也可以说成订阅式消费）

        消费者订阅对应的Topic,然后有消息服务器会推送给对应的消费者


        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("tudou");
    • 消费模式

      • 集群模式 默认情况下就是集群模式


        consumer.setMessageModel(MessageModel.CLUSTERING);


        如果消息接收者是个消息组，那么对应的消息组中的消费者一般会对应消息队列（例如队列有4个，消费组中有两个消费者，那么就会分别对应两个消费者队列，然后消息插入队列是顺序插入的就是一条消息插入到了第一条队列那么下一跳消息就会插入到第二条队列，这样也就做到了负载均衡）
      • 广播模式


        consumer.setMessageModel(MessageModel.BROADCASTING);


        广播模式: 每个消费者都会接受全量的消息, 所有消费者消费的数据都是一样的。一般用于对于消息需要多个其他业务进行处理，需要注意的是广播模式下消费者启动是不会消费启动之前的消息的，还有一点就是如果发送失败的消息也不会重试
    • 消息消费位置

      在指定消费的pos位置的时候, 会优先获取服务端记录的上次消费点, 所以该参数只有在服务端没有对应的消费者的记录的时候有效,一般情况是第一次启动的消费者有效

      defaultLitePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

      • CONSUME_FROM_FIRST_OFFSET

        从最开始的位置消费, 会消息该Topic下面所有的有效的数据, 过期的数据会删除掉
      • CONSUME_FROM_LAST_OFFSET

        1. 如果该Topic的数据都是最近的数据, 没有过期数据, 则从最开始的位置消费
        2. 如果该Topic的数据存在过期的数据, 则从最后的位置开始消费, 只会消费新加入的数据
      • CONSUME_FROM_TIMESTAMP

        根据指定的时间戳进行消费,配合

        consumer.setConsumeTimestamp("20200612083300");

        从指定的时间开始消费, 如果不指定, 则默认从半个小时前的数据开始消费
    • 消息确认

      • 拉式消息确认

        默认是自动提交的可以通过consumer.setAutoCommit(false);设置是否自动提交, 如果设置为手动提交, 需要使用consumer.commitSync();方法进行手动提交 对于未提交的操作, Topic中的订阅的偏移量是不会发生改变的, 下次消费的时候会继续消费改数据
      • 推式消息确认

        通过返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS状态表示消费成功 返回ConsumeConcurrentlyStatus.RECONSUME_LATER 表示消费不成功, 会放入到重试队列 默认重试采用服务端重试:  重试次数:16次

进阶篇

特殊消息处理

顺序消息

顺序消息指的是消费者能够按照消息发送的消息一次消费。实现原理就是指定队列

如果消息队列只有一个的情况下其实默认就是顺序的，应为在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。所以这里就衍生除了两种概念：

  • 全局有序

    全局有序比较好实现就是就是发动消息的时候指定消息队列就行，也就是所有的消息发送到一个队列就能保证消息的顺序

    MessageQueue messageQueue = new MessageQueue("queue_1","xzkjdeMac-mini.local",1);
SendResult result = producer.send(msg,messageQueue);
复制代码

    和平时发送消息不同的是需要添加一个指定队列

    可以发现只有queueId=1的队列才有消息

  • 局部有序

    举个例子：我们下个单冲创建订单->付款->发货这一系列的操作是需要按照绝对顺序的此时就可以根据订单Id取模获取到对应的队列Id(实现原理还是指定队列)，这样一个订单的一系列操作就会放到一个队列里面也就实现了局部有序.

延时消息

通过设置messageDelayLevel的等级来实现消息的延迟，默认的等级有18个等级1~18分别对应着1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，如下图所示：

这里需要注意的是RocketMQ不能随意设置延时时间，但是可以修改配置增加等级。

大概的原理是：延迟的消息不会立马放到消息队列中而是会放到一个Topic叫SCHEDULE_TOPIC_XXXX,然后延迟的等级对应这队列如下图所示：

时间到了之后再将消息转到对应的正式队列

对应代码如下： 

 // 4 创建消息对象
 Message msg = new Message("delay_1","testTag", ("hello tudou" + i).getBytes());
 msg.setDelayTimeLevel(6);
 // 5 发送消息
 SendResult result = producer.send(msg);
复制代码

过滤消息

  • Tag消息过滤


    defaultMQPushConsumer.subscribe("delay_1","TestA || TestB");

    如下图所示只消费了TestA和TestB标签的消息

  • SQL消息过滤

    RocketMQ只定义了一些基本语法来支持这个特性。你也可以很容易地扩展它。

    • 数值比较，比如： >，>=，<，<=，BETWEEN，=；
    • 字符比较，比如： =，<>，IN；
    • IS NULL 或者 IS NOT NULL；
    • 逻辑符号 AND，OR，NOT；

    常量支持类型为：

    • 数值，比如： 123，3.1415；
    • 字符，比如： ‘abc’，必须用单引号包裹起来；
    • NULL ，特殊的常量
    • 布尔值， TRUE 或 FALSE

    只有使用push模式的消费者才能用使用SQL92标准的sql语句，接口如下：


    public void subscribe(finalString topic, final MessageSelector messageSelector)

    一般使用SQL消息过滤之前是需要给消息赋值一些属性的然后用这些属性进行判断过滤 但是需要注意的是默认情况下是不支持属性过滤的需要通过配置参数开启在配置文件中配置对应的参数(conf/broker.conf)：enablePropertyFilter=true如下图所示：

    不配置会直接报错 这里需要注意的是如果换了配置启动命令中需要指定配置文件如下：


    nohup bin/mqbroker -c conf/broker.conf -n 127.0.0.1:9876 &


    否则走的还是默认的配置 最后可以跑一下代码测试：


    生产者：
     

    //1 创建一个生产者
DefaultMQProducer producer = new DefaultMQProducer("tudou1");
//2 指定NameServer的地址
producer.setNamesrvAddr("127.0.0.1:9876");
//3 启动生产者
producer.start();
for (int i = 0; i < 5; i++) {
    // 4 创建消息对象
    Message msg = new Message("delay_1","TestB", ("hello tudou" + i).getBytes());
    int score = 100 - i;
    msg.putUserProperty("score",String.valueOf(score));
    msg.putUserProperty("name","测试同学-"+i);

    // 5 发送消息
    SendResult result = producer.send(msg);

    System.out.println(result.getSendStatus());
    System.out.println("result.getMsgId() = " + result.getMsgId());
}
//6 关闭应用程序
producer.shutdown();
复制代码


    消费者：
     

    //1. 创建消费者对象
DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer("tudou_1");
//2. 设置NameServer地址
defaultMQPushConsumer.setNamesrvAddr("127.0.0.1:9876");
//3. 指定消费主题
defaultMQPushConsumer.subscribe("delay_1", MessageSelector.bySql("score > 98"));
//4. 冲那个位置开始消费
defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//5. 指定一个监听器，并发消费消息
defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt msg:list) {
            System.out.println("name =" + msg.getUserProperty("name") + "----- score =" +  msg.getUserProperty("score") + "----- msgBody =" +  new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
//6. 启动消费者
defaultMQPushConsumer.start();
复制代码


    消费结果：


    只消费了score大于98的消息

批量消息

批量消息每次发送消息不能超过4MB

集成SpringBoot

添加依赖 

  <dependency>
      <groupId>org.apache.rocketmq</groupId>
      <artifactId>rocketmq-spring-boot-starter</artifactId>
      <version>2.2.2</version>
  </dependency>
复制代码

生产者简单示例代码 

  @RestController
  public class TestController {

      @Resource
      private RocketMQTemplate rocketMQTemplate;

      @GetMapping("/send")
      public String test(String msg){
          SendResult result = rocketMQTemplate.syncSend("test",msg);
          return JSON.toJSONString(result);
      }

  }
复制代码

执行结果如下：

消费者简单示例代码 

  @Component
  @RocketMQMessageListener(
          consumerGroup = "test2",
          topic = "test")
  public class Custom implements RocketMQListener<String> {
      @Override
      public void onMessage(String s) {
          System.out.println("接收到的消息 ---- " + s);
      }
  }
复制代码

这里需要注意的是消费者组只能有一个否则报错

执行结果：

其他

其他的还有发送同步消息、异步消息、一次性消息这都比较简单我们可以看到明确的方法名如下图所示：

这里就不过多的演示了也就是简单的方法调用

  • 消费模式配置：

  • 延时消息


    注意：timeout并不是延时时间而是超时时间也就是说你这个消息发送最多不能超过这个时间

  • 设置标签

    在发送的消息Topic:Tag 中间使用冒号隔开


    rocketMQTemplate.convertAndSend("01-boot-hello:TagB",message,map);


    SendResult result = rocketMQTemplate.syncSend("test:Tag_A",msg);
  • 设置key

    //在消息的头信息中设置key
Message<?> message = MessageBuilder.withPayload(msg).setHeader(MessageConst.PROPERTY_KEYS, "testKey").build();
rocketMQTemplate.send("test_key", message);
复制代码
  • 自定义属性

    Map<String,Object> map=new HashMap<>();
//用户自定义属性
map.put("age", 18);
map.put("name", "tudou");
rocketMQTemplate.convertAndSend("test_tag:TagB",message,map);
复制代码

    也可以像上文设置key一样直接setHeader同样的key也可以使用这种方式设置

  • 过滤

  • 消息发送方式

    • 直接使用rocketMQTemplate
    • 使用DefaultMQProducer对象


      rocketMQTemplate.getProducer()

      获取该对象
    • 使用Spring的Message接口


      Message<String> message1 = MessageBuilder.withPayload(msg).build();

原理篇

架构设计

技术架构

上图是官网提供的一个架构图，从上图大概可以看出RocketMQ主要是由以下几个部分组成的：（上文也有提及）

  • Producer：消息发布的角色，支持分布式集群方式部署。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。

  • Consumer：消息消费的角色，支持分布式集群方式部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。

  • NameServer：NameServer是一个非常简单的Topic路由注册中心，其角色类似Dubbo中的zookeeper，支持Broker的动态注册与发现。主要包括两个功能：Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；路由信息管理，每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。NameServer通常也是集群的方式部署，各实例间相互不进行信息通讯。Broker是向每一台NameServer注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，Broker仍然可以向其它NameServer同步其路由信息，Producer,Consumer仍然可以动态感知Broker的路由的信息。

  • BrokerServer：Broker主要负责消息的存储、投递和查询以及服务高可用保证，为了实现这些功能，Broker包含了以下几个重要子模块。

    • Remoting Module：整个Broker的实体，负责处理来自clients端的请求。
    • Client Manager：负责管理客户端(Producer/Consumer)和维护Consumer的Topic订阅信息
    • Store Service：提供方便简单的API接口处理消息存储到物理硬盘和查询功能。
    • HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。

      所以在启动broker的时候会有个定时器定时的同步Master Broker 和 Slave Broker之间的数据
    • Index Service：根据特定的Message key对投递到Broker的消息进行索引服务，以提供消息的快速查询。

部署架构

RocketMQ 网络部署特点

  1. NameServer是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。(应为broker的心跳机制会想每一个NameServer发送信息，所以NameServer直接是不需要再同步消息的)

  2. Broker部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId 来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。 注意：当前RocketMQ版本在部署架构上支持一Master多Slave，但只有BrokerId=1的从服务器才会参与消息的读负载。

  3. Producer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic 服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。

  4. Consumer与NameServer集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，消费者在向Master拉取消息时，Master服务器会根据拉取偏移量与最大偏移量的距离（判断是否读老消息，产生读I/O），以及从服务器是否可读等因素建议下一次是从Master还是Slave拉取。

结合部署架构图，描述集群工作流程：

  1. 启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。

  2. Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。

  3. 收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。

  4. Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。

  5. Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

消息存储

消息存储流程

消息存储架构图中主要有下面三个跟消息存储相关的文件构成。

(1) CommitLog：消息主体以及元数据的存储主体，存储Producer端写入的消息主体内容,消息内容不是定长的。单个文件大小默认1G ，文件名长度为20位，左边补零，剩余为起始偏移量，比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1073741824；当第一个文件写满了，第二个文件为00000000001073741824，起始偏移量为1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件；（写入的消息会一直按照文件的顺序写入）。具体存储的位置是可以从配置信息中看到的如下图所示

(2) ConsumeQueue：消息消费队列，引入的目的主要是提高消息消费的性能，由于RocketMQ是基于主题topic的订阅模式，消息消费是针对主题进行的，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息Tag的HashCode值（如果使用tags过滤的话会使用到，在CommonQueue存储这个信息也比较好理解，就是过滤掉的信息就不需要再冲CommonL了og中查找读取）。consumequeue文件可以看成是基于topic的commitlog索引文件，故consumequeue文件夹的组织方式如下：topic/queue/file三层组织结构，具体存储路径为：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样consumequeue文件采取定长设计，每一个条目共20个字节，分别为8字节的commitlog物理偏移量、4字节的消息长度、8字节tag hashcode，单个文件由30W个条目组成，可以像数组一样随机访问每一个条目，每个ConsumeQueue文件大小约5.72M；

(3) IndexFile：IndexFile（索引文件）提供了一种可以通过key或时间区间来查询消息的方法。Index文件的存储位置是：HOME\store\indexHOME \store\indexHOME\store\index{fileName}，文件名fileName是以创建时的时间戳命名的，固定的单个IndexFile文件大小约为400M，一个IndexFile可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现HashMap结构，故rocketmq的索引文件其底层实现为hash索引。

在上面的RocketMQ的消息存储整体架构图中可以看出，RocketMQ采用的是混合型的存储结构，即为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。RocketMQ的混合型存储结构(多个Topic的消息实体内容都存储于一个CommitLog中)针对Producer和Consumer分别采用了数据和索引部分相分离的存储结构，Producer发送消息至Broker端，然后Broker端使用同步或者异步的方式对消息刷盘持久化，保存至CommitLog中。只要消息被刷盘持久化至磁盘文件CommitLog中，那么Producer发送的消息就不会丢失。正因为如此，Consumer也就肯定有机会去消费这条消息。当无法拉取到消息后，可以等下一次消息拉取，同时服务端也支持长轮询模式，如果一个消息拉取请求未拉取到消息，Broker允许等待30s的时间，只要这段时间内有新消息到达，将直接返回给消费端。这里，RocketMQ的具体做法是，使用Broker端的后台服务线程—ReputMessageService不停地分发请求并异步构建ConsumeQueue（逻辑消费队列）和IndexFile（索引文件）数据。

页缓存与内存映射


页缓存-PageCache

是OS对文件的缓存，用于加速对文件的读写。一般来说，程序对文件进行顺序读写的速度几乎接近于内存的读写速度，主要原因就是由于OS使用PageCache机制对读写访问操作进行了性能优化，将一部分的内存用作PageCache。对于数据的写入，OS会先写入至Cache内，随后通过异步的方式由pdflush内核线程将Cache内的数据刷盘至物理磁盘上。对于数据的读取，如果一次读取文件时出现未命中PageCache的情况，OS从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取。

如上图所示，此时需要读取9这个数据第一次用户态是回去访问页缓存如果此时有数据则直接命中，如果也缓存中没有数据则新开一页大小是4kb,然后此时也缓存在向硬盘拉取数据此时会将相邻的一些数据全部拉取过来存到也缓存中然后再将用户需要的数据返回。

写入的时候首先还是需要先判断页缓存中是否存在如果存在则修改否则新建一页添加数据对于被修改的页缓存，内核会定时把这些页缓存刷新到文件中。


所以为什么上文说到的使用顺序存储就是应为页缓存能极大的提交系统IO的速度

在RocketMQ中，ConsumeQueue逻辑消费队列存储的数据较少，并且是顺序读取，在page cache机制的预读取作用下，Consume Queue文件的读性能几乎接近读内存，即使在有消息堆积情况下也不会影响性能。而对于CommitLog消息存储的日志数据文件来说，读取消息内容时候会产生较多的随机访问读取，严重影响性能。如果选择合适的系统IO调度算法，比如设置调度算法为“Deadline”（此时块存储采用SSD的话），随机读的性能也会有所提升。

另外，RocketMQ主要通过MappedByteBuffer对文件进行读写操作。其中，利用了NIO中的FileChannel模型将磁盘上的物理文件直接映射到用户态的内存地址中（这种Mmap的方式减少了传统IO将磁盘文件数据在操作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销），将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率（正因为需要使用内存映射机制，故RocketMQ的文件存储都使用定长结构来存储，方便一次将整个文件映射至内存）。

Mmap补充：mmap是一种内存映射文件的方法，即将一个文件或者其它对象映射到进程的地址空间，实现文件磁盘地址和进程虚拟地址空间中一段虚拟地址的一一对映关系，正常情况下一个进程需要修改或者读取一个文件的数据是需要先从页缓存中读取过来就是需要内核态拷贝到用户态或者需要冲用户态拷贝到内核态，如果有了映射关系则禁止直接通过虚拟内存中拿到数据在内存中的地址可以直接通过指针读取和修改（当然无果内存中没有数据则会冲磁盘读取过来之后在进行读写）

消息刷盘

消息发送到Broker之后的存储流程如下图所示

先走到内存，然后为了数据的安全做持久化存储所以就在存到本地文件。 这里就牵扯到了同步刷盘和异步输盘：

  • 同步刷盘

    官方的图如下： 如上图所示最终存储到本地文件之后在收到ACK这种就属于同步刷盘

  • 异步刷盘

    官方的图如下： 如上图所示存储到内存就开始给ACK属于异步刷盘（这种方式会有消息丢失的风险）

负载均衡

Producer负载均衡

其实就是通过轮询发送到不同的队列上，让每个队列的消息数很平均

Consumer负载均衡

应为队列中的消息都是轮询插入的所以只需要消费者集群监听相同的队列就能实现负载均衡，这里需要注意的是队列数量要大于消费者服务的数量否则会出现某个消费服务没有消息可以消费

源码篇

消息中心启动流程

源码下载地址：

github.com/apache/rock…

NameServer启动流程


  1. 解析默认配置文件

    从加载的配置文件中可以发现namesrv使用的网络通信框架是Netty


  2. 根据执行命令参数执行对应操作


  3. 创建NameSrvController


  4. 进行初始化配置, 启动定时任务(定时检查Broker的状态)

     //定时任务查询broker状态
 this.scanExecutorService.scheduleAtFixedRate(NamesrvController.this.routeInfoManager::scanNotActiveBroker,
     5, this.namesrvConfig.getScanNotActiveBrokerInterval(), TimeUnit.MILLISECONDS);

 //定时加载kv配置信息
 this.scheduledExecutorService.scheduleAtFixedRate(NamesrvController.this.kvConfigManager::printAllPeriodically,
     1, 10, TimeUnit.MINUTES);
复制代码

    这里需要注意的是namesrv检测broker状态的机制并不是向broker发送消息，而是利用broker查看namesrv状态的机制来实现的，broker查看namesrv状态的机制是发送消息而且会发送当前时间，而namesrv会把每次broker过来的消息保存下来启动一个定时器每个几秒查看一下最新的broker时间和当前时间差值大于120秒则namesrv则会任务broker服务不可用


  5. 在RouteInfoManager中保存对应的配置信息


  6. 添加关闭线程池的钩子函数


  7. 最后启动服务

Broker启动流程


  1. 设置相关的参数（默认配置或者是根据命令加载配置文件)


  2. 创建BrokerController


  3. 进行初始化线程池配置, 启动一系列定时任务

    定时任务主要功能是：

    1. 定时往namesrv发送消息（心跳消息）
    2. 消费者组的一些持久化信息
    3. 通过定时任务完成延迟队列

  4. 绑定对应的关闭线程池的钩子函数


  5. 启动服务

消息发送流程

  1. 创建对象

    DefaultMQProducer


    这里主要是进行一些初始化的工作如下图所示

  2. 开始start方法

  3. 创建消息对象

  4. 发送下消息send


版权声明：本文为m0_73311735原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。