七点半的菜市场

large picture of the cover

三、RocketMQ快速入门

📦 Rocket版本 4.9.6

🏢 官方安装文档: https://rocketmq.apache.org/zh/docs/4.x/

💻 使用的系统是 ubuntu22.04

🏆 RocketMQ快速入门

⭐️ 生产者的基本概念

在生产者一章的基本概念包括消息，Tag，Keys，队列和生产者的介绍。

消息

RocketMQ 消息构成非常简单，如下图所示。

  • topic，表示要发送的消息的主题。
  • body 表示消息的存储内容
  • properties 表示消息属性
  • transactionId 会在事务消息中使用。
  • Tag: 不管是 RocketMQ 的 Tag 过滤还是延迟消息等都会利用 Properties 消息属性机制，这些特殊信息使用了系统保留的属性Key，设置自定义属性时需要避免和系统属性Key冲突。
  • Keys: 服务器会根据 keys 创建哈希索引，设置后，可以在 Console 系统根据 Topic、Keys 来查询消息，由于是哈希索引，请尽可能保证 key 唯一，例如订单号，商品 Id 等。

img

Message 可以设置的属性值包括：

字段名 默认值 必要性 说明
Topic null 必填 消息所属 topic 的名称
Body null 必填 消息体
Tags null 选填 消息标签，方便服务器过滤使用。目前只支持每个消息设置一个
Keys null 选填 代表这条消息的业务关键词
Flag 0 选填 完全由应用来设置，RocketMQ 不做干预
DelayTimeLevel 0 选填 消息延时级别，0 表示不延时，大于 0 会延时特定的时间才会被消费
WaitStoreMsgOK true 选填 表示消息是否在服务器落盘后才返回应答。

RocketMQ系统保留的属性Key集合有如下，需要在使用过程中避免：
TRACE_ON、MSG_REGION、KEYS、TAGS、DELAY、RETRY_TOPIC、REAL_TOPIC、REAL_QID、TRAN_MSG、PGROUP、MIN_OFFSET、MAX_OFFSET、BUYER_ID、ORIGIN_MESSAGE_ID、TRANSFER_FLAG、CORRECTION_FLAG、MQ2_FLAG、RECONSUME_TIME、UNIQ_KEY、MAX_RECONSUME_TIMES、CONSUME_START_TIME、POP_CK、POP_CK_OFFSET、1ST_POP_TIME、TRAN_PREPARED_QUEUE_OFFSET、DUP_INFO、EXTEND_UNIQ_INFO、INSTANCE_ID、CORRELATION_ID、REPLY_TO_CLIENT、TTL、ARRIVE_TIME、PUSH_REPLY_TIME、CLUSTER、MSG_TYPE、INNER_MULTI_QUEUE_OFFSET、_BORNHOST

Tag

Topic 与 Tag 都是业务上用来归类的标识，区别在于 Topic 是一级分类，而 Tag 可以理解为是二级分类。使用 Tag 可以实现对 Topic 中的消息进行过滤。

  • Topic：消息主题，通过 Topic 对不同的业务消息进行分类。
  • Tag：消息标签，用来进一步区分某个 Topic 下的消息分类，消息从生产者发出即带上的属性。

Topic 和 Tag 的关系如下图所示。

Tag

什么时候该用 Topic，什么时候该用 Tag？

可以从以下几个方面进行判断：

  • 消息类型是否一致：如普通消息、事务消息、定时（延时）消息、顺序消息，不同的消息类型使用不同的 Topic，无法通过 Tag 进行区分。
  • 业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使用不同的 Topic 进行区分；而同样是天猫交易消息，电器类订单、女装类订单、化妆品类订单的消息可以用 Tag 进行区分。
  • 消息优先级是否一致：如同样是物流消息，盒马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分。
  • 消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 Topic，则有可能会因为过长的等待时间而“饿死”，此时需要将不同量级的消息进行拆分，使用不同的 Topic。

总的来说，针对消息分类，您可以选择创建多个 Topic，或者在同一个 Topic 下创建多个 Tag。但通常情况下，不同的 Topic 之间的消息没有必然的联系，而 Tag 则用来区分同一个 Topic 下相互关联的消息，例如全集和子集的关系、流程先后的关系。

Keys

Apache RocketMQ 每个消息可以在业务层面的设置唯一标识码 keys 字段，方便将来定位消息丢失问题。 Broker 端会为每个消息创建索引（哈希索引），应用可以通过 topic、key 来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。

   // 订单Id
   String orderId = "20034568923546";
   message.setKeys(orderId);

队列

为了支持高并发和水平扩展，需要对 Topic 进行分区，在 RocketMQ 中这被称为队列，一个 Topic 可能有多个队列，并且可能分布在不同的 Broker 上。

MessageQueue

一般来说一条消息，如果没有重复发送（比如因为服务端没有响应而进行重试），则只会存在在 Topic 的其中一个队列中，消息在队列中按照先进先出的原则存储，每条消息会有自己的位点，每个队列会统计当前消息的总条数，这个称为最大位点 MaxOffset；队列的起始位置对应的位置叫做起始位点 MinOffset。队列可以提升消息发送和消费的并发度。

生产者

生产者（Producer）就是消息的发送者，Apache RocketMQ 拥有丰富的消息类型，可以支持不同的应用场景，在不同的场景中，需要使用不同的消息进行发送。比如在电商交易中超时未支付关闭订单的场景，在订单创建时会发送一条延时消息。这条消息将会在 30 分钟以后投递给消费者，消费者收到此消息后需要判断对应的订单是否已完成支付。如支付未完成，则关闭订单。如已完成支付则忽略，此时就需要用到延迟消息；电商场景中，业务上要求同一订单的消息保持严格顺序，此时就要用到顺序消息。在日志处理场景中，可以接受的比较大的发送延迟，但对吞吐量的要求很高，希望每秒能处理百万条日志，此时可以使用批量消息。在银行扣款的场景中，要保持上游的扣款操作和下游的短信通知保持一致，此时就要使用事务消息，下一节将会介绍各种类型消息的发送。

需要注意的是，生产环境中不同消息类型需要使用不同的主题，不要在同一个主题内使用多种消息类型，这样可以避免运维过程中的风险和错误。

⭐️ 消费者基础概念

消息通过生产者发送到某一个Topic，如果需要订阅该Topic并消费里面的消息的话，就要创建对应的消费者进行消费。在介绍消费者的使用方法之前，我们先介绍消费组、消费位点、推和拉等概念。

消费者与消费组

消息系统的重要作用之一是削峰填谷，但比如在电商大促的场景中，如果下游的消费者消费能力不足的话，大量的瞬时流量进入会后堆积在服务端。此时，消息的端到端延迟（从发送到被消费的时间）就会增加，对服务端而言，一直消费历史数据也会产生冷读。因此需要增加消费能力来解决这个问题，除了去优化消息消费的时间，最简单的方式就是扩容消费者。

但是否随意增加消费者就能提升消费能力？ 首先需要了解消费组的概念。在消费者中消费组的有非常重要的作用，如果多个消费者设置了相同的Consumer Group，我们认为这些消费者在同一个消费组内。

在 Apache RocketMQ 有两种消费模式，分别是：

  • 集群消费模式：当使用集群消费模式时，RocketMQ 认为任意一条消息只需要被消费组内的任意一个消费者处理即可。
  • 广播消费模式：当使用广播消费模式时，RocketMQ 会将每条消息推送给消费组所有的消费者，保证消息至少被每个消费者消费一次。

集群消费模式适用于每条消息只需要被处理一次的场景，也就是说整个消费组会Topic收到全量的消息，而消费组内的消费分担消费这些消息，因此可以通过扩缩消费者数量，来提升或降低消费能力，具体示例如下图所示，是最常见的消费方式。

集群消费模式

广播消费模式适用于每条消息需要被消费组的每个消费者处理的场景，也就是说消费组内的每个消费者都会收到订阅Topic的全量消息，因此即使扩缩消费者数量也无法提升或降低消费能力，具体示例如下图所示。

广播消费模式

负载均衡

集群模式下，同一个消费组内的消费者会分担收到的全量消息，这里的分配策略是怎样的？如果扩容消费者是否一定能提升消费能力？

Apache RocketMQ 提供了多种集群模式下的分配策略，包括平均分配策略、机房优先分配策略、一致性hash分配策略等，可以通过如下代码进行设置相应负载均衡策略

 consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragely());

默认的分配策略是平均分配，这也是最常见的策略。平均分配策略下消费组内的消费者会按照类似分页的策略均摊消费。

在平均分配的算法下，可以通过增加消费者的数量来提高消费的并行度。比如下图中，通过增加消费者来提高消费能力。

消费者扩容1

消费者扩容2

但也不是一味地增加消费者就能提升消费能力的，比如下图中Topic的总队列数小于消费者的数量时，消费者将分配不到队列，即使消费者再多也无法提升消费能力。

消费者扩容3

消费位点

消费位点

如上图所示，在Apache RocketMQ中每个队列都会记录自己的最小位点、最大位点。针对于消费组，还有消费位点的概念，在集群模式下，消费位点是由客户端提给交服务端保存的，在广播模式下，消费位点是由客户端自己保存的。一般情况下消费位点正常更新，不会出现消息重复，但如果消费者发生崩溃或有新的消费者加入群组，就会触发重平衡，重平衡完成后，每个消费者可能会分配到新的队列，而不是之前处理的队列。为了能继续之前的工作，消费者需要读取每个队列最后一次的提交的消费位点，然后从消费位点处继续拉取消息。但在实际执行过程中，由于客户端提交给服务端的消费位点并不是实时的，所以重平衡就可能会导致消息少量重复。

推、拉和长轮询

MQ的消费模式可以大致分为两种，一种是推Push，一种是拉Pull。

  • Push是服务端主动推送消息给客户端，优点是及时性较好，但如果客户端没有做好流控，一旦服务端推送大量消息到客户端时，就会导致客户端消息堆积甚至崩溃。
  • Pull是客户端需要主动到服务端取数据，优点是客户端可以依据自己的消费能力进行消费，但拉取的频率也需要用户自己控制，拉取频繁容易造成服务端和客户端的压力，拉取间隔长又容易造成消费不及时。

Apache RocketMQ既提供了Push模式也提供了Pull模式。

⭐️ 订阅一致性

订阅关系：一个消费者组订阅一个 Topic 的某一个 Tag，这种记录被称为订阅关系。

订阅关系一致：同一个消费者组下所有消费者实例所订阅的 Topic、Tag必须完全一致。如果订阅关系（消费者组名-Topic-Tag）不一致，会导致消费消息紊乱，甚至消息丢失。

⭐️ 快速入门发送(同步)/消费数据（push）

ps： 本来是要学 go的 rocketmq使用的，但是官方教程是 java，所以跟着官方教程来

1️⃣ 需要创建一个 Maven项目或者 springboot项目这里我就不做演示了，然后在 mave依赖中添加 如下依赖包

<!--RocketMQ-->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.4</version>
</dependency>
<!--用于测试-->
<dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.12</version>
</dependency>

2️⃣ 编写发送者代码（同步发送）

  1. 首先会创建一个producer。普通消息可以创建 DefaultMQProducer，创建时需要填写生产组的名称，生产者组是指同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。
  2. 设置 NameServer 的地址。Apache RocketMQ很多方式设置NameServer地址(客户端配置中有介绍)，这里是在代码中调用producer的API setNamesrvAddr进行设置，如果有多个NameServer，中间以分号隔开，比如"127.0.0.2:9876;127.0.0.3:9876"。
  3. 第三步是构建消息。指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤。
  4. 最后调用send接口将消息发送出去。同步发送等待结果最后返回SendResult，SendResult包含实际发送状态还包括SEND_OK（发送成功）, FLUSH_DISK_TIMEOUT（刷盘超时）, FLUSH_SLAVE_TIMEOUT（同步到备超时）, SLAVE_NOT_AVAILABLE（备不可用），如果发送失败会抛出异常。
package demo;

import export.export;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.junit.Test;


public class RocketMQ {
    @Test
    public void  Producer() throws Exception {
        //创建一个Producer
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("TestProducerGroup");
        //连接nameserver
        defaultMQProducer.setNamesrvAddr(export.NameSrv);
        //启动Producer
        defaultMQProducer.start();
        //创建消息
        Message message = new Message("AtestTopic","测试消息1".getBytes());
        //发送消息
        SendResult send = defaultMQProducer.send(message);
        //打印发送结果
        System.out.println("发送结果为："+send);
        //关闭Producer
        defaultMQProducer.shutdown();
    }
}

运行结果

image-20240801154018534

打开 dahsbord面板查看自动创建一个 topic（之前配置文件有写）

image-20240801154054397

查看状态，这表明这条消息在队列0中

image-20240801154132444

3️⃣ 编写消费者代码

首先需要初始化消费者，初始化消费者时，必须填写

ConsumerGroupName，同一个消费组的 ConsumerGroupName是相同的，这是判断消费者是否属于同一个消费组的重要属性。然后是设置 NameServer地址，这里与 Producer一样不再介绍。然后是调用 subscribe方法订阅 Topic，subscribe方法需要指定需要订阅的 Topic名，也可以增加消息过滤的条件，比如 TagA等，上述代码中指定*表示接收所有tag的消息。除了订阅之外，还需要注册回调接口编写消费逻辑来处理从 Broker中收到的消息，调用 registerMessageListener方法，需要传入 MessageListener的实现，上述代码中是并发消费，因此是 MessageListenerConcurrently的实现

其中，

msgs是从 Broker端获取的需要被消费消息列表，用户实现该接口，并把自己对消息的消费逻辑写在 consumeMessage方法中，然后返回消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS表示消费成功，或者表示 RECONSUME_LATER表示消费失败，一段时间后再重新消费

 /**
     * 创建消费者
     */
    @Test
    public void Consumer() throws Exception {
        //创建一个消费者
        DefaultMQPushConsumer defaultMQConsumer = new DefaultMQPushConsumer("TestConsumerGroup");
        //设置NameServer地址
        defaultMQConsumer.setNamesrvAddr(export.NameSrv);
        //订阅消息,构建方法使用topic和消息匹配规则，* 表示任何消息
        defaultMQConsumer.subscribe("AtestTopic","*");
        //注册回调接口来处理从Broker中收到的消息， MessageListenerConcurrently 是多线程消费，默认20个线程，可以参看consumer.setConsumeThreadMax()
        defaultMQConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.println("消息: "+msgs);
                System.out.println("上下文: "+context);
                // 返回消费的状态 如果是CONSUME_SUCCESS 则成功，若为RECONSUME_LATER则该条消息会被重回队列，重新被投递
                // 重试的时间为messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                // 也就是第一次1s 第二次5s 第三次10s  ....  如果重试了18次 那么这个消息就会被终止发送给消费者
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动Consumer
        defaultMQConsumer.start();
        //挂起jvm
        System.in.read();
    }

运行测试

image-20240801160730286

打开 dashboard查看

image-20240801160756407

site logo

Hi,Friend

© 2024 七点半的菜市场