同步消息#

• 特征：即时性较强，需要立即知道发送是否成功，例如短信、转账通知等

  原先Demo中的消息都是同步消息，即发送完立马可以得到result

异步消息#

• 特征：即时性较弱，无需立即知道是否发送成功，但后续必须要有result返回，例如订单中的某些信息

• 异步消息的Demo

  1
          //异步消息
  2
          producer.send(new Message("topic1", "Hello World".getBytes()),
  3
                  new SendCallback() {
  4
                      @Override
  5
                      //发送成功的回调方法
  6
                      public void onSuccess(SendResult sendResult) {
  7
                          System.out.println(sendResult);
  8
                      }
  9
  
  10
                      @Override
  11
                      //发送失败的回调方法
  12
                      public void onException(Throwable e) {
  13
                          System.out.println(e);
  14
                      }
  15
                  });

  展开收起

  异步消息在发送时需要传递一个回调函数。当异步消息发送出去后会继续向下执行业务代码，等到消息发送的结果返回时会调用回调函数

单向消息#

• 特征：无需知道是否发送成功，例如日志消息

• 单项消息Demo

  1
          //单向消息
  2
          producer.sendOneway(new Message("topic1", "Hello World".getBytes()));

延时消息#

• 特征：消息发送时并不直接发送到消息服务器，而是根据设定的等待时间到达，起到延时到达的缓冲作用

• 延时消息Demo

  1
          //延时消息
  2
          Message message = new Message("topic1", "Hello World".getBytes());
  3
          //设置延时等级
  4
          message.setDelayTimeLevel(1);
  5
          //设置具体延迟时间ms
  6
          message.setDelayTimeMs(10);
  7
          //设置具体延迟时间s
  8
          message.setDelayTimeSec(10);
  9
          //设置具体送达时间
  10
          message.setDeliverTimeMs(System.currentTimeMillis()+10*1000);
  11
          producer.send(message);

  以上四种延时方法选一种即可

批量消息#

• 特征：批处理发送消息，节约网络传输开销

• 批量消息的Demo

  1
          //批量消息
  2
          List<Message> msgs = new ArrayList<>();
  3
          Message m1 = new Message("topic1", "Hello World1".getBytes());
  4
          Message m2 = new Message("topic1", "Hello World2".getBytes());
  5
          Message m3 = new Message("topic1", "Hello World3".getBytes());
  6
  
  7
          msgs.add(m1);
  8
          msgs.add(m2);
  9
          msgs.add(m3);
  10
  
  11
          producer.send(msgs);

  实际上就是发送一个消息集合，批量消息还可以搭配异步

• 注意点

  1. 批量消息应该有相同的topic

  2. 相同的waitStoreMsgOK

  3. 批量消息不能是延时消息

  4. 消息内容总长度不能超过4M

     消息内容长度：topic+body+消息追加属性（key与value对应的字符串字节数）+日志（固定20字节）

tag过滤#

• 消费者在订阅队列时，第二个参数可以传递一个subExpression表达式，用来过滤同个Topic下的不同tag

  1
  consumer.subscribe("topic1", "*");
  2
  consumer.subscribe("topic1", "tag1||tag2");

  Tip：表达式只支持以上两种写法，不支持“tag*”这样的模糊匹配

SQL过滤#

• 消息可以设置追加属性

  1
  //Producer中添加属性
  2
  
  3
          //消息可以追加属性
  4
          Message m1 = new Message("topic1","tag1", "Hello World1".getBytes());
  5
          m1.putUserProperty("age", "17");
  6
          Message m2 = new Message("topic1", "tag1","Hello World2".getBytes());
  7
          m2.putUserProperty("age","20");
  8
  
  9
          SendResult res = producer.send(m1);
  10
          SendResult res1 = producer.send(m2);

• 在Consumer中可以利用类似SQL语法的语法去进行过滤

  1
          //按照SQL过滤
  2
          consumer.subscribe("topic1", MessageSelector.bySql("age > 18"));

• 注意

  SQL过滤默认是没有开启的，需要在broker.conf配置文件中添加enablePropertyFilter=true将其开启才可以正常使用该功能

  添加配置后重启broker需要添加以下参数

  1
  start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true -c ..\conf\broker.conf
  2
  # -c强制读取配置文件

Demo#

1. 导入依赖

   1
           <dependency>
   2
               <groupId>org.apache.rocketmq</groupId>
   3
               <artifactId>rocketmq-spring-boot-starter</artifactId>
   4
               <version>2.3.1</version>
   5
           </dependency>

2. 在application.yaml中配置命名服务器端口与生产者组

   1
   rocketmq:
   2
     name-server: localhost:9876
   3
     producer:
   4
       group: demo_producer

3. 注入RocketMQTemplate模板类（用于连接和断连）并编写demo

   1
       @Autowired
   2
       private RocketMQTemplate rocketMQTemplate;
   3
   
   4
       @GetMapping("/send")
   5
       public String send() throws InvalidProtocolBufferException {
   6
           String msg = new String("Hello MQWithSpringBoot");
   7
           //自动将消息转化为字节数组并向topic1发送消息
   8
           rocketMQTemplate.convertAndSend("topic1", msg);
   9
           //若向要指定tag，需要用:隔开
   10
           //rocketMQTemplate.convertAndSend("topic1:tag1", msg);
   11
   
   12
           return "success!";
   13
       }

4. 除此之外，convertAndSend 还可以序列化对象并传输

   注意：在网络中传输对象前必须对其进行序列化，对象类必须实现Serializable接口

   Serializable接口：标记型接口，无重写方法，用于标记该类允许被序列化

   1
   //User类
   2
   public class User implements Serializable {
   3
       private String name;
   4
       private String age;

   1
   //传输实体类
   2
           User u = new User("zhangsan",18);
   3
           rocketMQTemplate.convertAndSend("topic2", u);

5. 消费者的接收

   1. 将Consumer交给IOC容器管理
   2. 使用@RocketMQMessageListener注解指定topic、group、tag

   1
   @Service
   2
   @RocketMQMessageListener(topic = "topic2",consumerGroup = "group1")
   3
   //若想要指定tag
   4
   //@RocketMQMessageListener(topic = "topic1",consumerGroup = "group1",selectorExpression = "*")
   5
   public class DemoConsumer implements RocketMQListener<User> {
   6
   
   7
       @Override
   8
       public void onMessage(User message) {
   9
           //业务逻辑
   10
           System.out.println(message);
   11
       }
   12
   }

其他操作的集成#

• 发送同步消息

  1
      //同步发送
  2
          rocketMQTemplate.syncSend("topic",u);

• 异步发送

  1
          //异步发送
  2
          rocketMQTemplate.asyncSend("topic",u, new SendCallback() {
  3
              @Override
  4
              public void onSuccess(SendResult sendResult) {
  5
                  System.out.println(sendResult);
  6
              }
  7
  
  8
              @Override
  9
              public void onException(Throwable e) {
  10
                  System.out.println(e);
  11
              }
  12
          });

• 单向消息发送

  1
      rocketMQTemplate.sendOneWay("topic", u);

• 延迟发送

  1
      //延迟发送
  2
      rocketMQTemplate.syncSendDelayTimeSeconds("topic",u,20*1000);

• 批量发送

  1
          List<User> list = new ArrayList<>();
  2
          list.add(u);
  3
          rocketMQTemplate.syncSend("topic3", list);

• SQL过滤

  1
  @RocketMQMessageListener(topic = "topic2",consumerGroup = "group1",selectorType = SelectorType.SQL92,selectorExpression = "age>18")

• 改变消息接收模式

  1
  @RocketMQMessageListener(topic = "topic2",consumerGroup = "group1",messageModel = MessageModel.BROADCASTING)

消息顺序#

• 默认情况下，一个topic会生成四个消息队列

• 消息传输过程中可能会错乱

  • 消息错乱原因

    并发情况下，在一个订单创建的过程中会有其他订单也在创建，如此一来无法控制存入消息队列的顺序

    RocketMQ 只保证：同一个队列（MessageQueue）内的消息严格先进先出；不同队列之间消息完全无序。

    

  • 解决方法

    让同一笔订单的消息进入同一个队列，使得同一笔订单的不同消息在同一队列内排序。

    

    1. 生产者方：可以在存入消息的时候传入一个MessageQueueSelector(接口)的实现类 ，重写其中的队列选择方法，如此一来，每当要发送一个消息之前，都会根据重写的方法去选择存入哪个队列。我们只需自定义算法，确保同一笔订单的消息进入同一个队列即可

       在RocketMQ原生的API中MessageQueueSelector是在调用send()方法时传入的一个参数

       而在SpringBoot中，可以通过setMessageQueueSelector（） 来设置队列选择器

       1
               rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
       2
                   @Override
       3
             //mqs为存储队列的集合可以通过mqs.size()获取有几条队列  通过mqs.get(idx)获取指定队列
       4
             //msg为要发送的消息
       5
             //arg为专门用于路由运算的数据，如业务的key（orderid/userid等）
       6
                   public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
       7
                       return mqs.get(?);
       8
                   }
       9
               });

    2. 消费者方

       原生API：注册MessageListenerOrderly。

       在SpringBoot中被集成为注解中的consumerMode参数

       1
       @RocketMQMessageListener(topic = "topic2",consumerGroup = "group1",consumeMode = ConsumeMode.ORDERLY)

       一条队列只会被分配给一个消费组中的一个消费者

       MessageListenerOrderly与[RocketMQ Demo](#Rocket MQ Demo)中的MessageListenerConcurrently对比

       Concurrently在拿到队列时会直接开启多线程去处理队列信息，如此一来消息就会乱掉

       Orderly在拿到队列后会先获取锁，保证线程的串行执行，如此消息就不会乱了

事务消息#

• 概念

  

  一同执行本地数据库操作和MQ消息发送操作两个操作。

  1. 通过数据库事务的原子性保证 MQ消息发送的原子性
  2. 向本地数据库存入订单数据备份，以确保消息不丢失

  1. MQ发送half消息

     half消息：将消息完整地发给Broker，但是Broker并不会将其放入队列（消费者无法消费）

  2. Broker接收到half消息后返回OK

  3. 生产者操作数据库，开启事务

  4. 根据事务的成功或失败，向Broker发送提交或回滚。若事务成功Broker将消息加入队列；若事务失败Broker将消息删除

  • 事务补偿机制：若执行本地事务的过程中发生宕机、网络阻塞等情况，会导致Broker收不到提交或回滚信息，此时触发事务补偿机制 5. Broker向生产者发送未收到4的确认 6. 生产者检测本地事务状态 7. 检测后再向Broker发送事务的状态

• 实现

  1
          //1.事务消息使用的时事务消息生产者TransactionMQProducer
  2
          TransactionMQProducer producer = new TransactionMQProducer("group1");
  3
          producer.setNamesrvAddr("localhost:9876");
  4
  
  5
          //2.设置事务的监听
  6
          producer.setTransactionListener(new TransactionListener() {
  7
              @Override
  8
              //执行本地事务的逻辑（对应步骤3）
  9
              public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
  10
                  //将消息保存到本地数据库
  11
                  //....
  12
  
  13
                  //正常提交则返回COMMIT_MESSAGE状态
  14
                  //中途若抛出异常还可以用try catch包围并返回LocalTransactionState.ROLLBACK_MESSAGE 或 UNKNOW
  15
                  System.out.println("正常事务过程");
  16
                  return LocalTransactionState.COMMIT_MESSAGE;
  17
              }
  18
  
  19
              @Override
  20
              //抛异常时的检查（事务补偿）
  21
              public LocalTransactionState checkLocalTransaction(MessageExt msg) {
  22
                  System.out.println("事务补偿检查");
  23
  
  24
                  //通过sql去数据库中查询数据是否存入成功
  25
                  //若成功
  26
  //                return  LocalTransactionState.COMMIT_MESSAGE;
  27
                  //若失败
  28
                  return LocalTransactionState.ROLLBACK_MESSAGE;
  29
              }
  30
          });
  31
  
  32
          String msg = "ABC";
  33
          Message message = new Message("topic10", "tag1", msg.getBytes());
  34
          //发送事务消息
  35
          producer.sendMessageInTransaction(message,null);

  展开收起

消息的存储#

1. 生产者向MQ（Broker）发送消息
2. MQ接收到消息，将消息进行持久化，将数据刷盘存入磁盘中
3. MQ返回ACK给生产者，表示消息发送成功
4. MQ将消息推送给消费者
5. 消费者消费消息后返回ACK给MQ，表示该消息已被消费者消费
6. MQ将磁盘中的该条消息进行删除

• 注意

  第五步MQ在指定时间内未接收到ACK会认定消费失败，重新执行4、5、6

• RocketMQ快速读写的原因

  • RocketMQ的CommitLog文件是存储消息的文件，默认预先申请1G的空间，这样等到消息来时就可以直接进行顺序写入（而非随机写入），大大加快写的效率
  • 运用了“零拷贝”技术

• RocketMQ数据存储结构

  
  • commitLog存储了队列本身的数据
  • consumequeue消费者逻辑队列中存储了三种offset（偏移量），用于记录消息长度以及某一队列被哪个用户读到哪一条了。专门服务消费者
  • index索引，用于消息检索的哈希索引，只存hash值+CommitLog偏移，用于快速查找消息的位置

• RocketMQ刷盘机制

  • 同步刷盘
    1. 生产者发送消息到MQ，MQ接收到消息数据
    2. MQ挂起生产者发送消息的线程
    3. MQ将消息写入内存
    4. 内存数据写入硬盘
    5. 硬盘存储后返回SUCCESS
    6. 释放被挂起的生产者线程
    7. 发送ACK给生产者
  • 异步刷盘
    1. 生产者发送消息给MQ，MQ接收到消息
    2. MQ将消息写入内存
    3. MQ发送ACK给生产者
    4. 等到消息积累到一定的量，同一写入硬盘

高可用性#

• nameserver

  无状态**（集群后各nameserver间独立）**+全服务注册（所有的Broker会在每一台nameserver进行注册）

• 消息服务器（Broker）

  主从架构

• 消息生产者

  生产者将相同的topic绑定到多个group组，保障master挂掉后，其他master仍可以正常进行消息接收

• 消息消费者

  RocketMQ自身会根据master的压力确认是否由master承担消息读取的功能，当master繁忙时，自动切换由slave承担数据读取的工作。实现读写分离

负载均衡#

• Producer负载均衡

  • RocketMQ实现了不同broker集群中对同一topic对应消息队列的负载均衡

    即：若有一个topicA，在多个broker服务器中都有其队列存在，那么在Producer发送消息时，会尽量平均分配每一条队列的消息数量

    

• Consumer负载均衡

  • 平均分配

    为同一消费组内的不同消费者平均分配消息队列条数

  • 循环平均分配

    集群模式下，为不同消费者分担的队列如下图所示。如此一来，哪怕其中一个服务器宕机，也能确保各个消费者所承担的压力是相同的

    

消息重试#

• 消息消费后未正常返回消费成功的信息将启动消息重试机制

  • 消息重试机制

    • 顺序消息

      消息消费失败后，RocketMQ会自动进行消息重试**（间隔1s）**

      注意：应用会出现消息消费被阻塞的情况，因此，要对顺序消息的消费情况进行监控，避免消息阻塞导致消息一直重发

      

    • 无序消息

      无序消息包括普通消息、定时消息、延时消息、事务消息

      无序消息重试仅适用于负载均衡（集群）模型下的消息消费，不适用于广播模式下的消息消费

      为保障无序消息的消费，MQ 设定了合理的消息重试间隔时长

      

      默认尝试十六次，**每次的重试时间间隔都比前一次长 **若十六次都没成功，就不发送该消息

• 死信队列

  当消息重试达到一定次数后（默认16次），MQ将无法正常发送的消息称为死信消息存入死信队列

  • 死信队列特征

    • 归属某一个组（Group Id），而不归属 Topic，也不归属消费者
    • 一个死信队列中可以包含同一个组下的多个 Topic 中的死信消息
    • 死信队列不会进行默认初始化，当第一个死信出现后，此队列首次初始化

  • 死信队列中消息特征

  • 不会被再次重复消费

  • 死信队列中的消息有效期为 3 天，达到时限后将被清除

  • 死信处理

    1. 直接忽略不管
    2. 监控平台中，查找死信，获取死信的messageId，通过id对死信的精准消费

消息重复消费#

• 原因

  • 生产者发送了重复的消息
    • 网络闪断
    • 生产者宕机
  • 消费者重复消费
    • 网络闪断
    • broker重启
    • 消费者重启
    • 客户端扩容/缩容

• 解决

  • 使用业务id作为消息的key。
  • 消费者对key进行判定，未使用过则放行，使用过则抛弃