MyDistance

MyDistance

关注
发信
关注(28)粉丝(399)

RocketMQ 系列(三) 集成 SpringBoot

RocketMQ 系列(三) 集成 SpringBoot

前两篇文章介绍了 RocketMQ 基本概念与搭建，现在以它与 SpringBoot 的结合来介绍其基本的用法。

1、创建生产者

1.1、引入依赖

    <!-- RocketMQ -->
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.2.2</version>
    </dependency>

注意：rocketmq-spring-boot-starter要与RocketMQ的版本一致。

1.2、yaml 配置

application.yaml文件配置如下：

server:
  port: 9007
spring:
  application:
    name: rockmq-producer
rocketmq:
  # NameServer地址
  name-server: 192.168.0.17:9876
  producer:
    # 生产者组
    group: producer-group
    # 发送同步消息失败时，重试次数，默认是 2
    retry-times-when-send-failed: 2
    # 发送异步消息失败时，重试次数，默认是 2
    retry-times-when-send-async-failed: 2
    # 发送消息超时时间，默认是 3s
    send-message-timeout: 3000

1.3、编写发送消息接口

下面接口发送的为同步消息，即必须收到 RocketMQ 服务响应后才能进行下一步，否则一直阻塞。

@RequestMapping("/rocketmq")
@RestController
public class ProducerController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送同步消息
     */
    @RequestMapping("/syncSend")
    public void syncSend() {
        // 第一个参数指定Topic与Tag，格式: `topicName:tags`
        // 第二个参数，消息内容
        SendResult sendResult = rocketMQTemplate.syncSend("topicClean:tagTest", "syncMessage");
        System.out.println("发送同步消息结果：" + sendResult.toString());
    }
}

2、创建消费者

消费者的依赖同上面的生产者一样，同样是写下 yaml 文件配置。

2.1、yaml 配置

server:
  port: 9008
spring:
  application:
    name: rockmq-consumer
rocketmq:
  # NameServer地址
  name-server: 192.168.0.17:9876

2.2、编写消费者监听器

生产者发送消息到 broker 后，消费者通过监听的方式获取broker发送过来的消息。实现监听需要实现 RocketMQListener接口：

/**
 * 消费者监听器
 */
@Component
@RocketMQMessageListener(
        consumerGroup = "consumer-group",  //消费者组
        topic = "topicClean", //topic
        selectorExpression = "tagTest || tagB" //tag,可以有多个
)
public class ConsumerListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("接收消息:" + message);
    }
}

分析一下参数内容

  • topic 这个是必须指定的，否则没有消息来源。

  • consumerGroup 是消费者组，这个必须制定。一条消息只能被同一个消费者组里的一个消费者消费。

  • selectorExpression

    是用于消息过滤的，我们在生产的时候定义了tag内容，消费者可以指定消费某些tag的消息，具体策略如下：

    • 默认为 “*”，表示不过滤，消费此 topic 下所有消息
    • 配置为 “tagA”，表示只消费此 topic 下 TAG = tagA 的消息
    • 配置为 “tagTest || tagB”，表示消费此 topic 下 TAG = tagTest 或 TAG = tagB 的消息，以此类推

上面的@RocketMQMessageListener 注解的常用配置参数：

3、测试

首先第一步启动刚刚编写好的生产者及消费者服务。

调用生产者发送消息的接口/rocketmq/syncSend后，控制台返回结果 ,表示消息成功发送到 broker:

发送同步消息结果：SendResult [sendStatus=SEND_OK, msgId=7F000001178C18B4AAC288364E780000, offsetMsgId=7C471A0C00002A9F0000000000031086, messageQueue=MessageQueue [topic=topicClean, brokerName=broker-a, queueId=2], queueOffset=4]

查看 RocketMQ 控制台消息界面，也可以查询到刚刚发出来的消息：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

那么消费者是否成功的消费到消息了呢?这个我们暂时不清楚。

查看消费者控制台，很完美，消费者接收到了生产者的消息：

接收消息:syncMessage

同样，也可以查看 RocketMQ控制台消费者界面，上面我们确定的消费者组是consumer-group，点击查看消费详情，是能够看到成功地消费到了消息：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

生产者发送的消息成功被消费者消费，说明了基本的消息流程是没问题的。

上面我们发送的是同步消息，那这么说除了同步消息，还有其他哪几种消息阿？不了解，那我们就继续往下看。

4、消息类型

4.1、普通消息

上面发送的同步消息属于普通消息，普通消息就是 RocketMQ 中无特性的消息，包含了同步消息、异步消息、单步发送消息 3 种。

4.1.1、同步消息

同步消息是指消息发送方发出一条消息后，会在收到服务端返回响应之后才发下一条消息的通讯方式。

流程如下：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

应用场景：这种可靠性同步地发送方式应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。

示例代码：

@RequestMapping("/syncSend")
public void syncSend() {
    // 第一个参数指定Topic与Tag，格式: `topicName:tags`
    // 第二个参数，消息内容
    SendResult sendResult = rocketMQTemplate.syncSend("topicClean:tagTest", "syncMessage");
    System.out.println("发送同步消息结果：" + sendResult.toString());
}
4.1.2、异步消息

异步消息是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。RocketMQ 异步发送，需要实现异步发送回调接口（SendCallback）。消息发送方在发送了一条消息后，不需要等待服务端响应即可发送第二条消息，发送方通过回调接口接收服务端响应，并处理响应结果。

流程如下：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

应用场景：异步发送一般用于链路耗时较长，对响应时间较为敏感的业务场景，例如，视频上传后通知启动转码服务，转码完成后通知推送转码结果等。

示例代码：

@RequestMapping("/asyncSend")
public void asyncSend() {
    rocketMQTemplate.asyncSend("topicClean:tagTest", "asyncMessage", new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
            System.out.println("发送异步消息成功：" + sendResult.toString());
        }

        @Override
        public void onException(Throwable throwable) {
            System.out.println("发送异步消息失败：" + throwable.toString());
        }
    });
}
4.1.3、单步发送消息

发送⽅只负责发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不等待应答。

流程如下：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

应用场景：需要极快的响应速度，但不能保证可靠性。

示例代码：

@RequestMapping("/oneWaySend")
public void oneWaySend() {
    rocketMQTemplate.sendOneWay("topicClean:tagTest", "oneWayMessage");
}

4.2、顺序消息

顺序消息指的是，严格按照消息的发送顺序进行消费的消息(FIFO)。

默认情况下生产者会把消息以Round Robin轮询方式发送到不同的Queue分区队列；而消费消息时会从多个Queue上拉取消息，这种情况下的发送和消费是不能保证顺序的。

将消息仅发送到同一个Queue 中，消费时也只从这个 Queue 上拉取消息，就严格保证了消息的顺序性。

如何保证顺序：

  • 消息被发送时保持顺序
  • 消息被存储时保持和发送的顺序⼀致
  • 消息被消费时保持和存储的顺序⼀致

顺序消息分为全局有序消息、分区有序消息两种。

4.2.1、全局顺序消息

当发送和消费参与的 Queue 只有一个时所保证的有序是整个 Topic 中消息的顺序， 称为全局顺序，因为一个 Topic 对应只有一个 Queue, 所以会严重影响性能。

流程如下：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

在创建 Topic 时指定 Queue 的数量。有三种指定方式：

  • 在代码中创建 Producer 时，可以指定其自动创建的 Topic 的 Queue 数量
  • 在 RocketMQ 可视化控制台中手动创建 Topic 时指定 Queue 数量
  • 使用 mqadmin 命令手动创建 Topic 时指定 Queue 数量

只要将 Queue 的数量设置为 1 便可实现消息的有序存储。

4.2.2、分区顺序消息

对于指定的一个 Topic，所有消息根据 hashKey 进行区块分区。 同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费。

在电商业务场景中，一个订单的流程是：创建、付款、推送、完成。在加入 RocketMQ 后，一个订单会分别产生对于这个订单的创建、付款、完成消息，如果我们把所有消息全部送入到 RocketMQ 中的一个主题中，这里该如何实现针对一个订单的消息顺序性呢！

流程如下：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

要完成分区有序性，在生产者环节使用自定义的消息队列选择策略，确保订单号尾数相同的消息会被先后发送到同一个队列中(案例中主题有3个队列，生产环境中可设定成10个满足全部尾数的需求)，然后再消费端开启负载均衡模式，最终确保一个消费者拿到的消息对于一个订单来说是有序的。

生产者示例代码如下：

首先创建 order对象

public class Order {

    private long orderId;

    private String desc;

    public long getOrderId() {
        return orderId;
    }

    public Order setOrderId(long orderId) {
        this.orderId = orderId;
        return this;
    }

    public String getDesc() {
        return desc;
    }

    public Order setDesc(String desc) {
        this.desc = desc;
        return this;
    }

    @Override
    public String toString() {
        return "Order{" +
                "orderId=" + orderId +
                ", desc='" + desc + '\'' +
                '}';
    }
}

接着创建生产者分区顺序消息发送接口：

/**
     * 发送分区顺序消息
     */
@RequestMapping("/syncSendOrderly")
public void syncSendOrderly() {
    List<Order> orderList = new ArrayList<>();
    Order order = new Order();

    //订单1
    order.setOrderId(001).setDesc("创建");
    orderList.add(order);

    order = new Order().setOrderId(001).setDesc("付款");
    orderList.add(order);

    order = new Order().setOrderId(001).setDesc("完成");
    orderList.add(order);

    //订单2
    order = new Order().setOrderId(002).setDesc("创建");
    orderList.add(order);

    order = new Order().setOrderId(002).setDesc("付款");
    orderList.add(order);

    order = new Order().setOrderId(002).setDesc("完成");
    orderList.add(order);

    //订单3
    order = new Order().setOrderId(003).setDesc("创建");
    orderList.add(order);

    order = new Order().setOrderId(003).setDesc("付款");
    orderList.add(order);

    order = new Order().setOrderId(003).setDesc("完成");
    orderList.add(order);

    for (int i = 0; i < orderList.size(); i++) {
        //分区顺序消息
        //以orderId作为hashKey,一个 orderId 只会发送到一个 queue
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(
            "order-topic",
            "orderId:" +orderList.get(i).getOrderId() + ",orderMessage" + i,
            String.valueOf(orderList.get(i).getOrderId()));

        System.out.println(String.format("SendResult status:%s, queueId:%d, body:%s",
                                         sendResult.getSendStatus(),
                                         sendResult.getMessageQueue().getQueueId(),
                                         orderList.get(i).toString()));
    }
}

postman 调用接口测试，控制台输出结果如下:

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

结果显示相同的 orderId 被分配到同一个 queue,并且按照创建、付款、完成的步骤发送到了 broker，这里就实现了第一步：消息的有序存储。

顺序消费实际上有两个核心点，一个是生产者有序存储，另一个是消费者有序消费。

消费者示例代码如下：

/**
 * 消费者顺序消费监听器
 */
@Component
@RocketMQMessageListener(
        consumerGroup = "order--group",  //消费者组
        topic = "order-topic", //topic
        consumeMode = ConsumeMode.ORDERLY //消费模式：顺序消费，默认为并发消费
)
public class OrderConsumerListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("receive order message:" + message);
    }
}

注意：RocketMQMessageListener 的 consumeMode 属性默认为 ConsumeMode.CONCURRENTLY，实现顺序消息需要将类型需改为ConsumeMode.ORDERLY。

控制台显示消费结果如下：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

可以看到相同 orderId 的消息对应内容也是有序的。

4.3、延时消息

当消息写入到 broker 后，在指定的时长后才可被消费处理的消息，称为延时消息。

延时消息的延迟时长不支持随意时长的延迟，是通过特定的延迟等级来指定的。

messageDelayLevel = '1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h';

即，若指定的延时等级为 3，则表示延迟时长为 10s，即延迟等级是从 1 开始计数的。

当然，如果需要自定义的延时等级，可以通过在broker加载的 conf 配置中 新增如下配置（例如下面增加了 1 天这个等级 1d）。

messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d

延时消息实现原理如下：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

主要包含以下6个步骤：

  1. 修改消息 Topic 名称和队列信息

    RocketMQ Broker 端在存储生产者写入的消息时，首先都会将其写入到 CommitLog 中。之后根据消息中的 Topic 信息和队列信息，将其转发到目标 Topic 的指定队列(ConsumeQueue)中。

    由于消息一旦存储到 ConsumeQueue 中，消费者就能消费到，而延迟消息不能被立即消费，所以这里将Topic的名称修改为 SCHEDULE_TOPIC_XXXX，并根据延迟级别确定要投递到哪个队列下。同时，还会将消息原来要发送到的目标 Topic 和队列信息存储到消息的属性中。

  2. 转发消息到延迟主题 SCHEDULE_TOPIC_XXXX 的 CosumeQueue 中

    CommitLog 中的消息转发到 CosumeQueue中 是异步进行的。在转发过程中，会对延迟消息进行特殊处理，主要是计算这条延迟消息需要在什么时候进行投递。

  3. 延迟服务消费 SCHEDULE_TOPIC_XXXX 消息

    Broker 内部有一个 ScheduleMessageService 类，其充当延迟服务，主要是消费 SCHEDULE_TOPIC_XXXX 中的消息，并投递到目标 Topic 中。

    ScheduleMessageService 在启动时，其会创建一个定时器 Timer，并根据延迟级别的个数，启动对应数量的 TimerTask，每个 TimerTask 负责一个延迟级别的消费与投递。

  4. 将信息重新存储到 CommitLog 中

    在将消息到期后，需要投递到目标 Topic。由于在第一步已经记录了原来的 Topic 和队列信息，因此这里重新设置，再存储到 CommitLog 即可。

  5. 将消息投递到目标 Topic 中

  6. 消费者消费目标 Topic 中的数据。

应用场景： 在12306平台中，车票预订成功后就会发送一条延迟消息。这条消息将会在45分钟后投递给后台业务系统（Consumer），后台业务系统收到该消息后会判断对应的订单是否已经完成支付。如果未完成，则取消预订，将车票再次放回到票池；如果完成支付，则忽略。

示例代码如下：

    /**
     * 发送延时消息
     */
    @RequestMapping("/delaySend")
    public void delaySend() {
        //发送超时=3s,延时等级=3，延迟10s消费
        SendResult sendResult = rocketMQTemplate.syncSend("topicClean:tagTest",
                MessageBuilder.withPayload("延迟10s消息").build(), 3000, 3);
        System.out.println("发送延时消息：" + sendResult.toString());
    }

4.4、事务消息

RocketMQ 事务消息（Transactional Message）是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ 的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致。

事务消息发送分为两个阶段。第一阶段会发送一个半事务消息，半事务消息是指暂不能投递的消息，生产者已经成功地将消息发送到了 Broker，但是 Broker 未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，如果发送成功则执行本地事务，并根据本地事务执行成功与否，向 Broker 半事务消息状态（commit或者rollback），半事务消息只有 commit 状态才会真正向下游投递。如果由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，Broker 端会通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit或是Rollback）。这样最终保证了本地事务执行成功，下游就能收到消息，本地事务执行失败，下游就收不到消息。总而保证了上下游数据的一致性。

整个事务消息的详细交互流程如下图所示：

RocketMQ 系列(三) 集成 SpringBoot-LMLPHP

事务消息发送步骤如下：

  1. 生产者将半事务消息发送至 RocketMQ Broker。
  2. RocketMQ Broker 将消息持久化成功之后，向生产者返回 Ack 确认消息已经发送成功，此时消息暂不能投递，为半事务消息。
  3. 生产者开始执行本地事务逻辑。
  4. 生产者根据本地事务执行结果向服务端提交二次确认结果（Commit或是Rollback），服务端收到确认结果后处理逻辑如下：
    • 二次确认结果为 Commit：服务端将半事务消息标记为可投递，并投递给消费者。
    • 二次确认结果为 Rollback：服务端将回滚事务，不会将半事务消息投递给消费者。
  5. 在断网或者是应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达 MQ Server，经过固定时间后 MQ Server 将对该消息发起消息回查。
  6. 发送方收到消息回查后，需要检查对应消息的本地事务执行的最终结果。
  7. 发送方根据检查得到的本地事务的最终状态再次提交二次确认，MQ Server 仍按照步骤4对半消息进行操作。

应用场景：如用户发起转账后,交易状态短暂挂起,发送指令给银行,如果发起失败则不发送指令,发送成功后等待结果更新交易状态。

示例代码如下：

生产者：

private Logger logger = LoggerFactory.getLogger(getClass());

/**
     * 检查本地事务的状态
     */
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
    logger.info("start check Local rocketMQ transaction");

    RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;

    try {
        String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
        logger.info("check trans msg content：{}", jsonStr);
    } catch (Exception e) {
        //异常就回滚
        resultState = RocketMQLocalTransactionState.ROLLBACK;
    }
    return resultState;
}

说明：发送事务消息采用的是 sendMessageInTransaction 方法，返回结果为 TransactionSendResult 对象，该对象中包含了事务发送的状态、本地事务执行的状态等。

生产者监听器

发送事务消息除了生产者和消费者以外，我们还需要创建生产者的消息监听器，来监听本地事务执行的状态和检查本地事务状态。

/**
 * 事务消息监听器
 */
@RocketMQTransactionListener
public class TransactionMsgListener implements RocketMQLocalTransactionListener {
    private Logger logger = LoggerFactory.getLogger(getClass());


    /**
     * 执行本地事务
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
                                                                 Object arg) {
        logger.info("start invoke local rocketMQ transaction");
        RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;

        try {
            //处理业务
            String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
            logger.info("invoke msg content：{}", jsonStr);
            String UUID = (String) arg;
            logger.info("UUID:" + UUID);
        } catch (Exception e) {
            logger.error("invoke local mq trans error", e);
            resultState = RocketMQLocalTransactionState.UNKNOWN;
        }

        return resultState;
    }

    /**
     * 检查本地事务的状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        logger.info("start check Local rocketMQ transaction");

        RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;

        try {
            String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
            logger.info("check trans msg content：{}", jsonStr);
        } catch (Exception e) {
            //异常就回滚
            resultState = RocketMQLocalTransactionState.ROLLBACK;
        }
        return resultState;
    }
}

executeLocalTransaction 是半事务消息发送成功后，执行本地事务的方法，具体执行完本地事务后，可以在该方法中返回以下三种状态：

  • LocalTransactionState.COMMIT_MESSAGE：提交事务，允许消费者消费该消息
  • LocalTransactionState.ROLLBACK_MESSAGE：回滚事务，消息将被丢弃不允许消费。
  • LocalTransactionState.UNKNOW：暂时无法判断状态，等待固定时间以后 Broker 端根据回查规则向生产者进行消息回查。

checkLocalTransaction是由于二次确认消息没有收到，Broker 端回查事务状态的方法。回查规则：本地事务执行完成后，若 Broker 端收到的本地事务返回状态为 LocalTransactionState.UNKNOW，或生产者应用退出导致本地事务未提交任何状态。则 Broker 端会向消息生产者发起事务回查，第一次回查后仍未获取到事务状态，则之后每隔一段时间会再次回查。

消费者

/**
 * 消费者监听器
 */
@Component
@RocketMQMessageListener(
        consumerGroup = "consumer-group",  //消费者组
        topic = "topicClean", //topic
        selectorExpression = "tagTest || tagB" //tag
)
public class ConsumerListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println("接收消息:" + message);
    }
}

说明：事务消息的消费者与普通的消费者没有区别。

测试

调用事务消息接口，控制台打印日志如下：

l.p.listen.TransactionMsgListener        : start invoke local rocketMQ transaction
l.p.listen.TransactionMsgListener        : invoke msg content：this is transactionMessage
l.p.listen.TransactionMsgListener        : UUID:39030439-551f-407a-970b-a85f0671bfac
l.p.controller.ProducerController        : sendStatus:SEND_OK,localTransactionState:COMMIT_MESSAGE

通过日志我们可以看出，执行的流程与上述的一致，执行成功后，消息执行成功返回的结果为 SEND_OK，本地事务执行的状态为 COMMIT_MESSAGE。

异常测试

这里将修改executeLocalTransaction方法内容，当处理业务出现异常时，直接设置本地事务状态为ROLLBACK。

/**
     * 执行本地事务
     */
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
                                                             Object arg) {
    logger.info("start invoke local rocketMQ transaction");
    RocketMQLocalTransactionState resultState = RocketMQLocalTransactionState.COMMIT;

    try {
        //处理业务
        String jsonStr = new String((byte[]) msg.getPayload(), StandardCharsets.UTF_8);
        logger.info("invoke msg content：{}", jsonStr);
        //抛出异常
        int i = 1/0;
    } catch (Exception e) {
        logger.error("invoke local mq trans error", e);
        //设置事务状态为回滚
        resultState = RocketMQLocalTransactionState.ROLLBACK;
    }

    return resultState;
}

注意：

  • 若executeLocalTransaction 返回本地事务状态为UNKNOWN，Broker 端会进行事务回查，而事务回查执行的就是checkLocalTransaction方法。
  • 而如果executeLocalTransaction 返回本地事务状态为ROLLBACK，则直接丢弃准备发给消费者的消息，结束消息发送流程。

查看控制台，日志打印结果如下：

l.p.listen.TransactionMsgListener        : start invoke local rocketMQ transaction
l.p.listen.TransactionMsgListener        : invoke msg content：this is transactionMessage
l.p.listen.TransactionMsgListener        : invoke local mq trans error
l.p.controller.ProducerController        : sendStatus:SEND_OK,localTransactionState:ROLLBACK_MESSAGE

通过日志可以看出消息执行成功返回的结果为 SEND_OK，本地事务执行的状态为 ROLLBACK_MESSAGE。

本文演示了 Springboot 项目下 RocketMQ 消息的发送及消费流程，由最基本的同步消息举例讲解延伸到顺序消息、延时消息及事务消息这几种不同的消息类型。

想了解有关 RocketMQ 的更多知识点，且听下回（肝有点疼）。

参考资料：

09-06 18:50
Powered by LMLPHP ©2024 MyDistance 0.002651