RocketMQ 5.0 快速入门

RocketMQ 5.0

Apache RocketMQ 自诞生以来，因其架构简单、业务功能丰富、具备极强可扩展性等特点被众多企业开发者以及云厂商广泛采用。历经十余年的大规模场景打磨，RocketMQ 已经成为业内共识的金融级可靠业务消息首选方案，被广泛应用于互联网、大数据、移动互联网、物联网等领域的业务场景，现如今Apache RocketMQ也迭代到了5.0带来了很多新特性如重试机制、流控机制、消息清理机制、消费者负载均衡等

1、下载与安装RocketMQ 5.0

为了接近生产环境的开发，我们都是选择直接在Linux服务器上安装。

下载地址：https://rocketmq.apache.org/zh/download

注意：安装包分为二进制包和源码包，二进制包是已经编译好的可以直接运行，而源码包则需要编译后才能运行。

编译命令示例

$ unzip rocketmq-all-5.1.3-source-release.zip
$ cd rocketmq-all-5.1.3-source-release/
$ mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U
$ cd distribution/target/rocketmq-5.1.3/rocketmq-5.1.3

我们直接下载最新的二进制包就好了。

安装步骤如下

1、启动NameServer注册中心（存储 Broker 元信息）

# 解压
$ unzip rocketmq-all-5.1.3-bin-release.zip

解压后我们需要改一下启动脚本（如果服务器资源足够多可以忽略这一步）。

runserver.sh需要修改JVM内存的配置，此脚本默认从JVM申请的内存有4G（我们只是用来测试与学习服务器资源配置根本没有这么高），如下

# 以下为 runserver.sh 截取片段
# 无论走 if 还是 else -Xms和-Xmx的配置都是4g
# 所以我们要重新赋值这个 JAVA_OPT 变量
choose_gc_options()
{
    # Example of JAVA_MAJOR_VERSION value : '1', '9', '10', '11', ...
    # '1' means releases befor Java 9
    JAVA_MAJOR_VERSION=$("$JAVA" -version 2>&1 | awk -F '"' '/version/ {print $2}' | awk -F '.' '{print $1}')
    if [ -z "$JAVA_MAJOR_VERSION" ] || [ "$JAVA_MAJOR_VERSION" -lt "9" ] ; then
      JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
      JAVA_OPT="${JAVA_OPT} -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:-UseParNewGC"
      JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:${GC_LOG_DIR}/rmq_srv_gc_%p_%t.log -XX:+PrintGCDetails -XX:+PrintGCDateStamps"
      JAVA_OPT="${JAVA_OPT} -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=30m"
    else
      JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
      JAVA_OPT="${JAVA_OPT} -XX:+UseG1GC -XX:G1HeapRegionSize=16m -XX:G1ReservePercent=25 -XX:InitiatingHeapOccupancyPercent=30 -XX:SoftRefLRUPolicyMSPerMB=0"
      JAVA_OPT="${JAVA_OPT} -Xlog:gc*:file=${GC_LOG_DIR}/rmq_srv_gc_%p_%t.log:time,tags:filecount=5,filesize=30M"
    fi
}

重新定制内存（重新赋值JAVA_OPT变量）直接加在条件判断代码块后面即可。

# 重新定制内存
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
# JAVA_OPT="${JAVA_OPT} -XX:+UseG1GC -XX:G1HeapRegionSize=16m -XX:G1ReservePercent=25 -XX:InitiatingHeapOccupancyPercent=30 -XX:SoftRefLRUPolicyMSPerMB=0"
# JAVA_OPT="${JAVA_OPT} -Xlog:gc*:file=${GC_LOG_DIR}/rmq_srv_gc_%p_%t.log:time,tags:filecount=5,filesize=30M"

runbroker.sh也是需要修改JVM内存的配置，如下代码默认分配的是8g。

# 修改前
# JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g"
# 修改后
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m"

修改完后，我们就可以启动 RocketMQ 的 NameServer 了

# 启动 namesrv
$ nohup sh bin/mqnamesrv >mqnamesrv.log 2>&1 &
 
# 验证 namesrv 是否启动成功
$ tail -f -n 500 mqnamesrv.log
...
The Name Server boot success. serializeType=JSON, address 0.0.0.0:9876

# 或者是
$ tail -f ~/logs/rocketmqlogs/namesrv.log
2023-07-18 23:17:49 INFO NSScanScheduledThread - start scanNotActiveBroker
...

2、启动 Broker 消息存储中心和 Proxy 代理

修改/conf/broker.conf配置文件，这里主要是为了测试方便我们放开自动创建Topic的配置，加入以下配置

# 开启自动创建 Topic
autoCreateTopicEnable=true

Proxy组件是 RocketMQ 5.0 版本官方推荐的部署组件，详细说明可查看官方文档

https://rocketmq.apache.org/zh/docs/deploymentOperations/01deploy/

NameServer 成功启动后，我们启动 Broker 和 Proxy 。

# 启动（不使用代理）
nohup sh bin/mqbroker -n localhost:9876 >mqbroker.log 2>&1 &

# 启动 Broker+Proxy
$ nohup sh bin/mqbroker -n localhost:9876 --enable-proxy &

# 指定配置文件启动（broker默认使用的端口是10911，我们也可以在配置文件修改端口）
$ nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf --enable-proxy &

# 注意 --enable-proxy 开启代理后可能会报错
# java.io.IOException: Failed to bind to address 0.0.0.0:8080
# 当端口被占用时 broker/proxy 将无法启动
# 解决方案 https://blog.csdn.net/zooah212/article/details/127994243

# 验证是否启动成功
$ tail -f -n 500 nohup.out
The broker[suzhou-ydshp, 192.168.5.135:10911] boot success. serializeType=JSON and name server is localhost:9876

2、测试消息收发

创建消息发送的目标 Topic，RocketMQ 5.0 版本需要提前创建，例如：

# 可以通过 mqadmin 命令创建
# 注意 TestTopic 是topic名称
$ sh bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster

create topic to 192.168.5.135:10911 success.
TopicConfig [topicName=TestTopic, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false, attributes={}]

1、在IDEA中创建一个Java工程，并引入以下依赖

<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-test -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client-java -->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client-java</artifactId>
    <version>5.0.5</version>
</dependency>

2、创建发送消息的程序并运行

@SpringBootTest
public class DemoApplicationTest {
    private static final Logger logger = LoggerFactory.getLogger(DemoApplicationTest.class);

    @Test
    public void test() throws ClientException {
        // 接入点地址，需要设置成 Proxy 的地址和端口列表，一般是xxx:8081
        String endpoint = "192.168.5.135:8081";
        // 消息发送的目标Topic名称，需要提前创建。
        String topic = "TestTopic";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
        ClientConfiguration configuration = builder.build();

        // 初始化Producer时需要设置通信配置以及预绑定的Topic
        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();

        // 普通消息发送
        Message message = provider.newMessageBuilder()
                .setTopic(topic)
                // 设置消息索引键，可根据关键字精确查找某条消息
                .setKeys("messageKey")
                // 设置消息Tag，用于消费端根据指定Tag过滤消息
                .setTag("messageTag")
                // 消息内容实体（byte[]）
                .setBody("hello rocketMQ".getBytes())
                .build();
        try {
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(message);
            logger.info("send message successfully, messageId={}", sendReceipt.getMessageId());
        } catch (ClientException e) {
            logger.error("failed to send message", e);
        }
        // 关闭
        producer.close();
    }
}

3、创建订阅消息程序并运行。

Apache RocketMQ 支持SimpleConsumer和PushConsumer两种消费者类型，可以选择任意一种方式订阅消息。这里主要介绍PushConsumer。

@Test
public void pushConsumerTest() throws Exception {
    ClientServiceProvider provider = ClientServiceProvider.loadService();
    // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081
    String endpoint = "192.168.5.135:8081";
    ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoint)
            .build();
    // 订阅消息的过滤规则，表示订阅所有Tag的消息
    String tag = "*";
    FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
    // 为消费者指定所属的消费者分组，Group需要提前创建
    String consumerGroup = "TestGroup";

    // 指定需要订阅哪个目标Topic，Topic需要提前创建
    String topic = "TestTopic";
    // 初始化 PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系
    PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            // 设置消费者分组
            .setConsumerGroup(consumerGroup)
            // 设置预绑定的订阅关系
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            // 设置消费监听器
            .setMessageListener(messageView -> {
                // 处理消息并返回消费结果
                logger.info("consume message successfully, messageId={}", messageView.getMessageId());
                // 消息内容处理
                ByteBuffer body = messageView.getBody();
                String message = StandardCharsets.UTF_8.decode(body).toString();
                body.flip();
                logger.info("message body={}", message);
                return ConsumeResult.SUCCESS;
            }).build();
    Thread.sleep(Long.MAX_VALUE);
    // 如果不需要再使用 PushConsumer，可关闭该实例。
    pushConsumer.close();
}

3、新版rocketmq-dashboard搭建

rocketmq-dashboard是由 rocketmq-console 升级而来，整体UI风格更加简洁，新增了很多新功能。支持多种部署方式如 docker 镜像部署，源码手动编译与部署等。

搭建过程可参考如下文章

文章地址：https://blog.csdn.net/m0_46357847/article/details/130476251

官方文档（使用说明）：https://rocketmq.apache.org/zh/docs/deploymentOperations/04Dashboard

4、原生API的使用

RocketMQ的原生API在GitHub上都能找到使用案例，这里主要是列举了常用的API。

GitHub地址：

https://github.com/apache/rocketmq/tree/develop/example/src/main/java/org/apache/rocketmq/example

4.1、异步消息发送

RocketMQ 5.0提供了新的API来支持发送异步消息，例如

@Test
public void AsyncProducerTest() throws ClientException, InterruptedException {
    String endpoint = "192.168.5.135:8081"; // Proxy的地址和端口
    ClientConfigurationBuilder builder = ClientConfiguration.newBuilder()
            .setEndpoints(endpoint);
    ClientConfiguration configuration = builder.build();

    String topic = "async_topic";
    ClientServiceProvider provider = ClientServiceProvider.loadService();
    Producer producer = provider.newProducerBuilder()
            .setTopics(topic)
            .setClientConfiguration(configuration)
            .build();

    // 设置消息
    Message message = provider.newMessageBuilder()
            .setTopic(topic)
            .setKeys("async_key")
            .setTag("async_tag")
            .setBody("hello rocketMQ this is a async message".getBytes())
            .build();

    // 异步消息发送
    CompletableFuture<SendReceipt> completableFuture = producer.sendAsync(message); // 返回一个task编排工具
    // 回调处理
    completableFuture.whenComplete((sendReceipt, exception) -> {
        if (null != exception) {
            exception.printStackTrace();
        } else {
            logger.info("send message successfully, messageId={}", sendReceipt.getMessageId());
            try {
                producer.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    });
    Thread.sleep(Long.MAX_VALUE);
}

同步接收消息，前面我们已经介绍过了PushConsumer的使用，这里使用SimpleConsumer来同步消费消息，例如

// 使用SimpleConsumer消费消息
@Test
public void AsyncConsumerTest() throws Exception {
    String endpoint = "192.168.5.135:8081";
    ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoint)
            .build();

    String consumerGroup = "test_group";
    FilterExpression filterExpression = new FilterExpression("*", FilterExpressionType.TAG);
    String topic = "async_topic";

    ClientServiceProvider provider = ClientServiceProvider.loadService();
    SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            .setConsumerGroup(consumerGroup)
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            .setAwaitDuration(Duration.ofSeconds(30))
            .build();
    List<MessageView> messageViewList = null;
    try {
        messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
        messageViewList.forEach(messageView -> {
            // 消息内容处理
            ByteBuffer body = messageView.getBody();
            String message = StandardCharsets.UTF_8.decode(body).toString();
            body.flip();
            logger.info("message body={}", message);

            //消费处理完成后，需要主动调用ACK提交消费结果
            try {
                simpleConsumer.ack(messageView);
            } catch (ClientException e) {
                e.printStackTrace();
            }
        });
    } catch (ClientException e) {
        // 如果遇到系统流控等原因造成拉取失败，需要重新发起获取消息请求
        e.printStackTrace();
    }
    // Thread.sleep(Long.MAX_VALUE);
    // simpleConsumer.close();
}

先执行异步消息发送代码，在执行消费消息代码，我们再来看消息的收发情况。

4.2、异步消费消息

新版本的RocketMQ也支持异步消费消息，例如：

@Test
public void AsyncConsumerTest() throws Exception {
    String endpoint = "192.168.5.135:8081";
    ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoint)
            .build();

    String consumerGroup = "test_group";
    FilterExpression filterExpression = new FilterExpression("*", FilterExpressionType.TAG);
    String topic = "async_topic";

    ClientServiceProvider provider = ClientServiceProvider.loadService();
    SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            .setConsumerGroup(consumerGroup)
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            .setAwaitDuration(Duration.ofSeconds(30))
            .build();
    List<MessageView> messageViewList = null;
    // 同步消费
    // messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    // 异步消费
    CompletableFuture<List<MessageView>> completableFutureList = simpleConsumer.receiveAsync(10, Duration.ofSeconds(30));
    messageViewList = completableFutureList.get();
    if (CollectionUtils.isEmpty(messageViewList)) {
        return;
    }
    messageViewList.forEach(messageView -> {
        // 消息内容处理
        ByteBuffer body = messageView.getBody();
        String message = StandardCharsets.UTF_8.decode(body).toString();
        body.flip();
        logger.info("message body={}", message);

        //消费处理完成后，需要主动调用ACK提交消费结果
        try {
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });
    Thread.sleep(Long.MAX_VALUE);
    // simpleConsumer.close();
}

4.3、事务消息发送

由官方文档得知事务消息为 Apache RocketMQ 中的高级特性消息，分布式系统调用的特点为一个核心业务逻辑的执行，同时需要调用多个下游业务进行处理。因此，需要保证核心业务和多个下游业务的执行结果完全一致。

以电商交易场景为例：

支付订单这一核心操作会涉及到

• 主分支订单系统状态更新（由未支付变更为支付成功）
• 下游物流发货（新增待发货物流记录，创建订单物流记录）
• 积分变更（变更用户积分，更新用户积分表）
• 购物车状态清空（清空购物车，更新用户购物车记录）
• 等多个子系统的变更

需要做事务控制，已经很明显了。而在这之前一直使用XA分布式事务方案，XA分布式事务方案将四个调用分支封装成包含四个独立事务分支的大事务。基于XA分布式事务的方案可以满足业务处理结果的正确性，但最大的缺点是多分支环境下资源锁定范围大，并发度低，随着下游分支的增加，系统性能会越来越差。

因此新版本 Apache RocketMQ 解决了一系列分布式性能问题、多个下游分支一致性等问题，推出了分布式事务消息功能，与本地事务（数据库事务）进行绑定，使其拥有提交、回滚和统一协调的能力，并且具备高性能、可扩展、业务开发简单的优势。

更多介绍请参考官方文档（事务消息处理流程、事务消息生命周期等）：
https://rocketmq.apache.org/zh/docs/featureBehavior/04transactionmessage

事务消息使用示例：

1、新增一个支持事务消息的topic官方文档也明确指出 NORMAL 类型Topic不支持TRANSACTION类型消息，生产消息会报错。

# 注意 transaction_topic 是topic名称
$ ./bin/mqadmin updatetopic -n localhost:9876 -t transaction_topic -c DefaultCluster -a +message.type=TRANSACTION

create topic to 192.168.5.135:10911 success.
TopicConfig [topicName=transaction_topic, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false, attributes={+message.type=TRANSACTION}]

2、发送事务消息

@Test
public void transactionProviderTest() throws Exception {
    String endpoint = "192.168.5.135:8081"; // Proxy的地址和端口
    ClientConfiguration configuration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoint).build();

    String topic = "transaction_topic";
    // 构造事务生产者：事务消息需要生产者构建一个事务检查器，用于检查确认异常事务的中间状态
    ClientServiceProvider provider = ClientServiceProvider.loadService();
    Producer producer = provider.newProducerBuilder()
            .setTopics(topic)
            .setClientConfiguration(configuration)
            // 设置事务检查器（不知道啥时候触发的。。。遇到了在研究一下吧）
            .setTransactionChecker(messageView -> {
                // 事务检查器一般是根据业务的ID去检查本地事务是否正确提交还是回滚，此处以订单ID属性为例
                // 在订单表找到了这个订单，说明本地事务插入订单的操作已经正确提交；如果订单表没有订单，说明本地事务已经回滚
                String orderId = messageView.getProperties().get("orderId");
                if (!StringUtils.hasText(orderId)) {
                    // 没有业务ID直接回滚
                    return TransactionResolution.ROLLBACK;
                }
                // 检查本地事务是否提交
                String order = getOrderById(orderId);
                logger.info("check transaction start order={} [orderId={}]", order, orderId);
                if (!StringUtils.hasText(order)) {
                    // 本地事务没有正常提交直接回滚
                    return TransactionResolution.ROLLBACK;
                }
                // 通过业务ID查询到了对应的记录说明本地事务已经正常提交了
                // 消息事务也正常提交
                return TransactionResolution.COMMIT;
            }).build();

    // 开启事务分支
    Transaction transaction;
    try {
        transaction = producer.beginTransaction();
    } catch (ClientException e) {
        e.printStackTrace();
        // 事务分支开启失败则直接退出
        return;
    }

    // 构建事务消息
    Message message = provider.newMessageBuilder()
            .setTopic(topic)
            .setKeys("transaction_key")
            .setTag("transaction_tag")
            .addProperty("orderId", UUID.randomUUID().toString()) // 一般事务消息都会设置一个本地事务关联的唯一ID，用来做本地事务回查的校验
            .setBody("hello rocketMQ this is a transaction message".getBytes(StandardCharsets.UTF_8))
            .build();

    // 发送事务消息
    SendReceipt sendReceipt;
    try {
        sendReceipt = producer.send(message, transaction);
    } catch (ClientException e) {
        e.printStackTrace();
        // 事务消息发送失败，事务可以直接退出并回滚
        return;
    }
    /**
     * 执行本地事务，并确定本地事务结果
     * 1. 如果本地事务提交成功，则提交消息事务
     * 2. 如果本地事务提交失败，则回滚消息事务
     * 3. 如果本地事务未知异常，则不处理，等待事务消息回查
     */
    boolean localTransactionOk = doLocalTransaction();
    if (localTransactionOk) {
        try {
            transaction.commit();
        } catch (ClientException e) {
            // 业务可以自身对实时性的要求选择是否重试，如果放弃重试，可以依赖事务消息回查机制进行事务状态的提交
            e.printStackTrace();
        }
    } else {
        try {
            transaction.rollback();
        } catch (ClientException e) {
            // 建议记录异常信息，回滚异常时可以无需重试，依赖事务消息回查机制进行事务状态的提交
            e.printStackTrace();
        }
    }
    Thread.sleep(Long.MAX_VALUE);
}

// 模拟订单查询服务用来确认订单事务是否提交成功
private static String getOrderById(String orderId) {
    return "order";
}

// 模拟本地事务执行结果
private static boolean doLocalTransaction() {
    return true;
}

4.4、顺序消息发送

RocketMQ 也支持按顺序发送与消费消息。在某些特定的场景异构系统之间需要保持强一致的状态同步，上游的事件变更需要按照顺序传递到下游进行处理。

官方文档也给出了两个应用场景。

场景1：撮合交易

以证券、股票交易撮合场景为例，对于出价相同的交易单，坚持按照先出价先交易的原则，下游处理订单的系统需要严格按照出价顺序来处理订单。

场景2：数据实时增量同步

以数据库变更增量同步场景为例，上游源端数据库按需执行增删改操作，将二进制操作日志作为消息，通过 Apache RocketMQ 传输到下游搜索系统，下游系统按顺序还原消息数据，实现状态数据按序刷新。如果是普通消息则可能会导致状态混乱，和预期操作结果不符，基于顺序消息可以实现下游状态和上游操作结果一致。

顺序消息实现原理：

1、顺序消息需要设置消息组。Apache RocketMQ 顺序消息的顺序关系通过消息组（MessageGroup）判定和识别，发送顺序消息时需要为每条消息设置归属的消息组，相同消息组的多条消息之间遵循先进先出的顺序关系，不同消息组、无消息组的消息之间不涉及顺序性。

2、需要保证消息生产的顺序性！也就是说生产者必须保证单一性（单一生产者），不同生产者分布在不同的系统，即使设置相同的消息组，不同生产者之间产生的消息也无法判定其先后顺序。

3、消息发送必须串行发送。Apache RocketMQ 生产者客户端支持多线程安全访问，但如果生产者使用多线程并行发送，则不同线程间产生的消息将无法判定其先后顺序。

4、消费者类型为PushConsumer时， Apache RocketMQ 保证消息按照存储顺序一条一条投递给消费者。消费者类型为SimpleConsumer，则消费者有可能一次拉取多条消息。因此需要自己控制拉取逻辑以保证顺序消费。

5、重试次数必须有限！Apache RocketMQ 顺序消息投递仅在重试次数限定范围内，即一条消息如果一直重试失败，超过最大重试次数后将不再重试，跳过这条消息消费，不会一直阻塞后续消息处理。

6、顺序消息仅支持使用MessageType为FIFO的主题，即顺序消息只能发送至类型为顺序消息的主题中，发送的消息的类型必须和主题的类型一致。

代码实现：

1、创建FIFO主题

# Apache RocketMQ 5.0版本 官方推荐我们使用mqadmin工具创建主题
# -c 集群名称 
# -t Topic名称 
# -n nameserver地址 
# -o 创建顺序消息
# -a +message.type 指定主题容纳的消息类型(TRANSACTION、FIFO、DELAY、NORMAL如未指定该属性默认是NORMAL)

# 注意 FIFOTopic 主题名称
$ ./bin/mqadmin updateTopic -c DefaultCluster -t FIFOTopic -o true -n 127.0.0.1:9876 -a +message.type=FIFO

create topic to 192.168.5.135:10911 success.
set cluster orderConf. isOrder=true, orderConf=[broker-a:8]
TopicConfig [topicName=FIFOTopic, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=true, attributes={}]

2、生产者代码实现

@Test
public void fifoProvider() throws ClientException {
    String endpoint = "192.168.5.135:8081"; // Proxy的地址和端口
    ClientConfiguration configuration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoint).build();

    String topic = "FIFOTopic";
    String messageGroup = "fifo_group";

    ClientServiceProvider provider = ClientServiceProvider.loadService();
    Producer producer = provider.newProducerBuilder()
            .setClientConfiguration(configuration)
            .setTopics(topic)
            .build();
	// 发送5条消息并给消息做个顺序标记
    for (int i = 0; i < 5; i++) {
        Message message = provider.newMessageBuilder()
                .setTopic(topic)
                .setKeys("fifo_key")
                .setTag("fifo_tag")
                .setMessageGroup(messageGroup) // 设置顺序消息的排序分组，该分组尽量保持离散，避免热点排序分组
                .setBody(String.format("hello rocketMQ this is a fifo message_%s", i).getBytes())
                .build();
        try {
            SendReceipt sendReceipt = producer.send(message);
            logger.info("send message successfully, messageId={}", sendReceipt.getMessageId());
        } catch (ClientException e) {
            logger.error("failed to send message", e);
        }
    }
}

3、消费者代码实现

@Test
public void fifoConsumer() throws Exception {
    String endpoint = "192.168.5.135:8081"; // Proxy的地址和端口
    ClientConfiguration configuration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoint).build();

    String topic = "FIFOTopic";
    String messageGroup = "fifo_group";

    ClientServiceProvider provider = ClientServiceProvider.loadService();
    // 消费顺序消息时，需要确保当前消费者分组是顺序投递模式，否则仍然按并发乱序投递
    String tag = "*";
    FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

    // 消息消费监听器（不要写成 Lambda 表达式形式的）
    // Lambda 相当于是一个匿名函数，是无序性的
    MessageListener messageListener = new MessageListener() {
        @Override
        public ConsumeResult consume(MessageView messageView) {
            // 处理消息并返回消费结果
            // logger.info("consume message successfully, messageId={}", messageView.getMessageId());
            // 消息内容处理
            ByteBuffer body = messageView.getBody();
            String message = StandardCharsets.UTF_8.decode(body).toString();
            body.flip();
            logger.info("message body={}", message);
            return ConsumeResult.SUCCESS;
        }
    };
    // 使用PushConsumer消费顺序消息，只需要在消费监听器处理即可
    PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(configuration)
            .setConsumerGroup(messageGroup)
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            .setMessageListener(messageListener).build();
    Thread.sleep(Long.MAX_VALUE);
    // pushConsumer.close();
}

4、先启动消费者进行监听，然后在启动生产者发送消息。

5、查看消费者消费消息的顺序

4.5、延时/定时消息发送

由官方文档得知定时消息和延时消息本质上是相同的（因此下文可以统一使用定时消息描述），都是服务端根据消息设置的定时时间在某一固定时刻将消息投递给消费者消费。

定时消息也是RocketMQ非常重要的特性之一，基于传统的数据库定时调度方案，性能不高，实现复杂，而基于Apache RocketMQ的定时消息可以封装出多种类型的定时触发器，适应性强更易扩展。因此定时消息常见的应用场景如下：

场景1：分布式定时调度

在分布式定时调度触发、任务超时处理等场景，需要实现精准、可靠的定时事件触发。使用 Apache RocketMQ 的定时消息可以简化定时调度任务的开发逻辑，实现高性能、可扩展、高可靠的定时触发能力。

在分布式定时调度场景下，需要实现各类精度的定时任务，例如每天5点执行文件清理，每隔2分钟触发一次消息推送等需求都可以使用 Apache RocketMQ 实现。

场景2：任务超时处理

以电商交易场景为例，订单下单后暂未支付，此时不可以直接关闭订单，而是需要等待一段时间后才能关闭订单。使用 Apache RocketMQ 定时消息可以实现超时任务的检查触发。

基于定时消息的超时任务处理具备如下优势：

1. 精度高、开发门槛低：基于消息通知方式不存在定时阶梯间隔。可以轻松实现任意精度事件触发，无需业务去重。

2. 高性能可扩展：传统的数据库扫描方式较为复杂，需要频繁调用接口扫描，容易产生性能瓶颈。 Apache RocketMQ 的定时消息具有高并发和水平扩展的能力。

实现原理及生命周期https://rocketmq.apache.org/zh/docs/featureBehavior/02delaymessage#功能原理

代码实现：

1、创建DELAY类型的Topic

定时消息仅支持在 MessageType为DELAY的主题内使用，即定时消息只能发送至类型为定时消息的主题中，发送的消息的类型必须和主题的类型一致。

cd bin
sh mqadmin updateTopic -c DefaultCluster -t delay_topic -n 127.0.0.1:9876 -a +message.type=DELAY

# 创建成功后，会输出一下信息
create topic to 192.168.5.135:10911 success.
TopicConfig [topicName=delay_topic, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false, attributes={+message.type=DELAY}]

2、生产者代码实现

@Test
public void delayProvider() throws ClientException {
    String endpoint = "192.168.5.135:8081"; // Proxy的地址和端口
    ClientConfiguration configuration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoint).build();

    String topic = "delay_topic";

    ClientServiceProvider provider = ClientServiceProvider.loadService();
    Producer producer = provider.newProducerBuilder()
            .setClientConfiguration(configuration)
            .setTopics(topic)
            .build();

    // 延迟时间为10分钟之后的Unix时间戳。
    Long deliverTimeStamp = System.currentTimeMillis() + 10L * 60 * 1000;

    Message message = provider.newMessageBuilder()
            .setTopic(topic)
            .setKeys("delay_key")
            .setTag("delay_tag")
            .setDeliveryTimestamp(deliverTimeStamp)
            .setBody("hello rocketMQ this is a delay message".getBytes())
            .build();
    try {
        SendReceipt sendReceipt = producer.send(message);
        logger.info("send message successfully, messageId={}", sendReceipt.getMessageId());
    } catch (ClientException e) {
        logger.error("failed to send message", e);
    }
}

3、消费者代码实现

@Test
public void consumer() throws Exception {
    String endpoint = "192.168.5.135:8081"; // Proxy的地址和端口
    ClientConfiguration configuration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoint).build();

    String topic = "delay_topic";
    String messageGroup = "delay_group";

    ClientServiceProvider provider = ClientServiceProvider.loadService();
    String tag = "*";
    FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

    // 消息消费监听器
    MessageListener messageListener = messageView -> {
        // 处理消息并返回消费结果
        // logger.info("consume message successfully, messageId={}", messageView.getMessageId());
        // 消息内容处理
        ByteBuffer body = messageView.getBody();
        String message = StandardCharsets.UTF_8.decode(body).toString();
        body.flip();
        logger.info("message body={}", message);
        return ConsumeResult.SUCCESS;
    };
    PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(configuration)
            .setConsumerGroup(messageGroup)
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            .setMessageListener(messageListener).build();
    Thread.sleep(Long.MAX_VALUE);
    // pushConsumer.close();
}

4、先启动消费者进行监听

5、启动生产者发送延时消息

6、等待10分钟，看消费者消费情况

4.6、原生API小结

关于原生API的使用简单了解到这里就足够了，实际开发中我们很少使用原生API实现业务功能，基本上都是集成至SpringBoot或者是配合SpringCloud一起使用，官方文档也还有很多知识比如消息发送重试和流控机制、消息清理机制等可自行学习。

5、SpringBoot 3.0集成RocketMQ 5.0

5.1、RocketMQAutoConfiguration自动配置类分析

1、引入依赖坐标

SpringBoot版本我们使用3.0版本的，依赖坐标如下：

<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-parent -->
<!-- 所有SpringBoot项目都要继承spring-boot-starter-parent -->
<parent>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-parent</artifactId>
  <version>3.0.7</version>
  <relativePath/> <!-- lookup parent from repository -->
</parent>

到目前为止 Apache 官方对 rocketmq-spring-boot-starter 场景启动器还没有更新到 3.0 版本，因此我们还是用2.0版本的场景启动器，如下：

<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-spring-boot-starter -->
<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-spring-boot-starter</artifactId>
  <version>2.2.3</version>
</dependency>

2、编写 application.yaml 配置文件

rocketmq:
  name-server: 192.168.5.135:9876
  producer:
    group: dist-test # 生产者组
  pull-consumer: # pull模式消费者
    group: test
    topic: test

关于配置文件RocketMQ配置说明：

/*
1.生产者和消费者的配置必须配置其中一个（要么配置生产者、要么配置消费者、要么都配置）
2、配置生产者时name-server/producer.group不能为空
3、配置消费者时name-server/pull-consumer.roup/pull-consumer.topic不能为空
*/

否则 RocketMQTemplate 会注入失败！！！

源码分析（只截取了关键的代码）：

...
// 1. matchIfMissing = true 如果没有在application.yaml设置该属性，则默认为条件符合
// 也就是说即使配置文件没有配置rocketmq.name-server配置项依然会扫描此自动配置类
@ConditionalOnProperty(prefix = "rocketmq", value = {"name-server"}, matchIfMissing = true)
...
public class RocketMQAutoConfiguration implements ApplicationContextAware {
  	...
    // 2.默认创建的生产者（重要条件：name-server和producer.group必须配置才会注入容器中）
  	@Bean({"defaultMQProducer"})
    @ConditionalOnMissingBean({DefaultMQProducer.class})
    @ConditionalOnProperty(
        prefix = "rocketmq",
        value = {"name-server", "producer.group"}
    )
    public DefaultMQProducer defaultMQProducer(RocketMQProperties rocketMQProperties) {
      ...
    }
  
  	// 3.默认创建的消费者（重要条件：name-server/pull-consumer.roup/pull-consumer.topic必须配置才会注入容器中）
    @Bean({"defaultLitePullConsumer"})
    @ConditionalOnMissingBean({DefaultLitePullConsumer.class})
    @ConditionalOnProperty(
        prefix = "rocketmq",
        value = {"name-server", "pull-consumer.group", "pull-consumer.topic"}
    )
    public DefaultLitePullConsumer defaultLitePullConsumer(RocketMQProperties rocketMQProperties) throws MQClientException {
      ...
    }
  
  	// 4.RocketMQTemplate
  	@Bean(destroyMethod = "destroy")
  	// 4.1 核心就在这个注解（可以点进去 ProducerOrConsumerPropertyCondition 看一下）
  	// 4.2 如果默认创建的生产者和消费者都没有注入容器，则不会满足条件，也就不会向容器中注入RocketMQTemplate
    @Conditional({RocketMQAutoConfiguration.ProducerOrConsumerPropertyCondition.class})
    @ConditionalOnMissingBean(name = {"rocketMQTemplate"})
    public RocketMQTemplate rocketMQTemplate(RocketMQMessageConverter rocketMQMessageConverter) {
      ...
    }
  	...
}

3、编写主程序

我们都知道SpringBoot中，默认集成的消息中间件是RabbitMQ，默认情况下不会加载RocketMQ的自动配置类，如下图所示

并且 SpringBoot 3.0 自动配置类信息书写的文件由 META-INF/spring.factories 变更为 META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 。

因此在 SpringBoot 3.0 中引入的 RocketMQ 场景启动器是2.0版本的话，需要手动导入 RoketMQ 的自动配置类，如下

@SpringBootApplication
@ImportAutoConfiguration({RocketMQAutoConfiguration.class}) // 手动导入 RocketMQ 的自动配置类
public class TestApplication {
    public static void main(String[] args) {
        SpringApplication.run(TestApplication.class, args);
    }
}

5.2、消息收发的基本使用

1、生产者

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

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @GetMapping("/send")
    public R send() {
    	/**
         * 普通消息：只负责发送无需等待响应
         * 参数1：topic:tag tag可省略不写
         * 参数2：Object类型的消息体
         * 消费监听器：一般配合着 RocketMQListener<T> 使用
         */
        rocketMQTemplate.convertAndSend("test", "hello world");
        return R.ok();
    }

	@GetMapping("/send2")
    public R send2() {
    	/**
    	 * 普通消息：等待消费者响应
    	 * 参数信息和上面一样
    	 * 消费者监听器：一般配合着 RocketMQReplyListener<T, R> 使用
    	 */
        String res = rocketMQTemplate.sendAndReceive("test", "hello RocketMQ", String.class);
        return R.success(res);
    }
}

2、消费者

// RocketMQReplyListener<T, R> 是一个接口，需要返回值的监听器，两个泛型分别是接收消息的类型和返回值类型
// RocketMQListener<T> 无需返回值 T 为接收消息的类型
@Component
@RocketMQMessageListener(topic = "test", consumerGroup = "test_consumer")
public class TestRocketMQMessageListener implements RocketMQReplyListener<String, String> {
    private static final Logger log = LoggerFactory.getLogger(TestRocketMQMessageListener.class);

    @Override
    public String onMessage(String s) {
        log.info("接收到RocketMQ消息[topic={}] ======> {}", "test", s);
        return SendStatus.SEND_OK.name();
    }
}

3、测试

监听信息

5.3、Tag过滤消息

Tag标签过滤为精准字符串匹配，过滤规则设置格式如下：

/*
1.单Tag匹配：过滤表达式为目标Tag。表示只有消息标签为指定目标Tag的消息符合匹配条件，会被发送给消费者。

2.多Tag匹配：多个Tag之间为或的关系，不同Tag间使用两个竖线（||）隔开。例如，Tag1||Tag2||Tag3，表示标签为Tag1或Tag2或Tag3的消息都满足匹配条件，都会被发送给消费者进行消费。

3.全部匹配（默认规则）：使用星号（*）作为全匹配表达式。表示主题下的所有消息都将被发送给消费者进行消费。
*/

1、生产者

@RestController
@RequestMapping("/rocketmq")
public class SendController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @GetMapping("/send")
    public R send() {
        String destination = "test:pay";
        /*rocketMQTemplate.send(destination, MessageBuilder.withPayload()
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build());*/
        rocketMQTemplate.send(destination, MessageBuilder.withPayload("这是一个支付消息").build());
        return R.ok();
    }
}

2、消费者

@Component
@RocketMQMessageListener(
        topic = "test",
        consumerGroup = "test_consumer",
        selectorExpression = "pay || order" // tag是pay和order都能监听到
)
public class TestRocketMQMessageListener implements RocketMQListener<String> {
    private static final Logger log = LoggerFactory.getLogger(TestRocketMQMessageListener.class);

    @Override
    public void onMessage(String message) {
        log.info("接收到RocketMQ消息[topic={}] ======> {}", "test", message);
    }
}

5.3、SQL过滤消息

SQL属性过滤是 Apache RocketMQ 提供的高级消息过滤方式，通过生产者为消息设置的属性（Key）及属性值（Value）进行匹配。生产者在发送消息时可设置多个属性，消费者订阅时可设置SQL语法的过滤表达式过滤多个属性。

例如：

// Map<String, Object> headers 就是生产者可设置的key-value属性
public void convertAndSend(D destination, Object payload, @Nullable Map<String, Object> headers) throws MessagingException {
    this.convertAndSend(destination, payload, headers, (MessagePostProcessor)null);
}

// 或者是
MessageBuilder.setHeader()

SQL属性过滤使用SQL92语法作为过滤规则表达式，语法规范如下：

注意使用SQL92过滤消息，需要手动配置Broker支持SQL92，否则应用会启动失败（The broker does not support consumer to filter message by SQL92）

# 关闭 broker 在 broker.conf 中添加如下配置
enablePropertyFilter = true
# 重启broker
nohup ./mqbroker -n localhost:9876 -c ../conf/broker.conf >../mqbroker.log 2>&1 &

使用示例：

1、生产者

@RestController
@RequestMapping("/rocketmq")
public class SendController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @GetMapping("/send")
    public R send() {
        String destination = "test:pay";
        /*rocketMQTemplate.send(destination, MessageBuilder.withPayload()
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build());*/
        rocketMQTemplate.send(destination, MessageBuilder
                .withPayload("这是一个支付消息")
                .setHeader("name", "alex")
                .build());
        return R.ok();
    }
}

2、消费者

@Component
@RocketMQMessageListener(
        topic = "test",
        consumerGroup = "test_consumer",
        selectorType = SelectorType.SQL92,
        selectorExpression = "name IS NOT NULL AND name = 'alex'"
)
public class TestRocketMQMessageListener implements RocketMQListener<String> {
    private static final Logger log = LoggerFactory.getLogger(TestRocketMQMessageListener.class);

    @Override
    public void onMessage(String message) {
        log.info("接收到RocketMQ消息[topic={}] ======> {}", "test", message);
    }
}

5.4、事务消息

事务消息相关知识前面我们也有介绍到，需要补充的是事务消息的执行流程：

/*
1、正常事务消息发送与提交阶段
	1.生产者发送一个半消息给RocketMQ（半消息是指消费者暂时不能消费的消息）
	2.服务端响应消息写入结果，半消息发送成功
	3.开始执行本地事务
	4.根据本地事务的执行状态执行Commit或者Rollback操作

2、事务信息的补偿流程（补偿阶段主要是用于解决生产者在发送Commit或者Rollback操作时发生超时或失败的情况）
	1.如果RocketMQ长时间没收到本地事务的执行状态会向生产者发起一个确认回查的操作请求
	2.生产者收到确认回查请求后，检查本地事务的执行状态
	3.根据检查后的结果执行Commit或者Rollback操作
	
3、RocketMQ是如何实现事务消息提交前消费者不可见呢？
	1.事务消息相对普通消息最大的特点就是一阶段发送的消息对用户是不可见的，也就是说消费者不能直接消费。
	2.这里RocketMQ的实现方法是原消息的主题与消息消费队列，然后把主题改成RMQ_SYS_TRANS_HALF_TOPIC 这样由于消费者没有订阅这个主题，所以不会被消费
————————————————
版权声明：本文为CSDN博主「HGW689」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/m0_49183244/article/details/129169326
*/

代码示例：

1、生产者

@RestController
@RequestMapping("/rocketmq")
public class SendController {
    private static final Logger log = LoggerFactory.getLogger(SendController.class);
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @GetMapping("/send")
    public R send() {
        String destination = "transaction_topic:tx";
        for (int i = 1; i < 6; i++) {
            Message<String> message = MessageBuilder.withPayload(String.format("事务消息%s", i))
                    .setHeader("orderId", i)
                    .build();
            Map<String, Object> params = new HashMap<>();
            // 这里的 orderId 可以放在setHeader()
            // 也可以放在 params 建议放在header上 事务监听器上 检查事务状态 方法没有params参数

            // 发送事务消息
            TransactionSendResult res = rocketMQTemplate.sendMessageInTransaction(destination, message, params);
            log.info("msgId = {} , sendStatus = {}", res.getMsgId(), res.getSendStatus());
        }
        return R.ok();
    }
}

2、生产者事务监听器

@Component
@RocketMQTransactionListener
public class TestRocketMQLocalTransactionListener implements RocketMQLocalTransactionListener {
    private static final Logger log = LoggerFactory.getLogger(TestRocketMQLocalTransactionListener.class);

    // 生产者数据库事务逻辑写在这里
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        MessageHeaders headers = message.getHeaders();
        // 本地事务id
        String transactionId = (String) headers.get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);
        String orderId = (String) headers.get("orderId");
        log.info("local transaction start transactionId = {} [orderId = {}]", transactionId, orderId);
        if (!StringUtils.hasLength(orderId)) {
            // 直接回滚
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        int id = Integer.parseInt(orderId);
        // ===== 本地事务开始 =====
        // 执行保存 orderService.saveOrder(order) 执行本地事务
        // ===== 本地事务结束 =====

        // 模拟本地事务执行成功(偶数)和失败(奇数)
        if (id % 2 == 0) {
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            // 假设这里的失败是本地事务还在执行(还不确定提交还是回滚)
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    /**
     * 检查事务状态（监听回查请求）
     * 1.当成产者执行本地事务发生故障或者是返回 UNKNOWN 状态,要保证这条消息最终被消费,RocketMQ会像服务端发送回查请求,确认本地事务的执行状态
     * 2.不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还是无法得知事务状态，RocketMQ默认回滚该消息
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        MessageHeaders headers = message.getHeaders();
        String transactionId = (String) headers.get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);
        String orderId = (String) headers.get("orderId");
        log.info("check transaction start transactionId = {} [orderId = {}]", transactionId, orderId);
        if (!StringUtils.hasLength(orderId)) {
            // 直接回滚
            return RocketMQLocalTransactionState.ROLLBACK;
        }

        // 查询 orderService.getOrderById(id)
        // 如果能查到则说明本地事务执行成功 返回rocketMQLocalTransactionState.COMMIT
        // 反之则说明本地事务还在执行或者是出现故障

        // 这里统一模拟为出现故障，不让消费者消费消息
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}

3、消费者

@Component
@RocketMQMessageListener(
        topic = "transaction_topic",
        consumerGroup = "test_consumer",
        selectorExpression = "tx"
)
public class TestRocketMQMessageListener implements RocketMQListener<String> {
    private static final Logger log = LoggerFactory.getLogger(TestRocketMQMessageListener.class);

    @Override
    public void onMessage(String message) {
        log.info("接收到RocketMQ消息[topic={}] ======> {}", "test", message);
    }
}

4、最终只有偶数的会被消费掉

学会普通消息、事务消息、消息过滤已经可以满足大多数场景了。后面还顺序消息、延迟消息、批量消息等都比较简单，用法也是大同小异，遇到了自行研究下即可。

6、SpringCloudStream集成RocketMQ 5.0

6.1、相关概念介绍

消息驱动 Spring Cloud Stream 是企业和微服务项目使用最多的消息驱动的框架，因为 Spring Cloud Stream 可以屏蔽底层消息中间件的差异，降低切换成本，统一消息的变成模型。就像 Hibernate 一样统一提供sesstionAPI，无需关注各数据库厂商的语法。

注意：Spring Cloud Stream 在新版本中官方已经弃用相关注解了（@EnableBinding/@Input/@Output等），取而代之的是通过更加灵活的函数式编程实现，简称SCS框架。

除此之外，新版本全面使用函数式编程，带来了很多新的概念，但是部分核心概念依然保留如：

1、Destination Binders：负责提供与外部消息传递系统集成的组件，完美的实现了应用程序与消息中间件细节之间的隔离

2、Destination Bindings：外部消息传递系统和最终用户提供的应用程序代码（生产者/消费者）之间的桥梁。

3、Message：生产者和消费者使用的规范数据结构，用于与目的地绑定器通信（从而通过外部消息传递系统与其他应用程序通信）。

6.2、如何使用新版本SpringCloudStream

了解相关概念后，我们再来看怎么使用新版本 Spring Cloud Stream ？

引入依赖坐标

<!-- https://mvnrepository.com/artifact/com.alibaba.cloud/spring-cloud-starter-stream-rocketmq -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
    <version>2022.0.0.0</version>
</dependency>

函数式接口初体验：

1、通过函数式接口来声明Binding

@Configuration
public class ScsConfiguration {

    // Supplier声明一个消息生产者
    @Bean
    public Supplier<String> source() {
        String message = "send a message";
        // 发送消息
        return () -> {
            System.out.println(String.format("我是生产者：已发送了一条消息 =====> %s", message));
            return message;
        };
    }

    // Consumer声明一个消息消费者
    @Bean
    public Consumer<String> sink() {
        return message -> System.out.println(String.format("我是消费者：接收到一条消息 =====> %s", message));
    }
}

2、修改配置文件，引入声明的生产者和消费者

spring:
  cloud:
    stream:
      function:
        definition: source;sink # bean的名称

3、这样应用启动后就会源源不断的生产消息和消费消息

上面例子是自动产生消息，实际开发中很少有应用场景，我们都需要手动发送消息，手动发送消息又该如何实现？

手动发送消息：

Spring Cloud Stream 新版本提供了新的组件 StreamBridge 用于手动发送消息，StreamBridge的API如下：

// StreamBridge 实现了 StreamOperations 接口
public interface StreamOperations {
    boolean send(String bindingName, Object data);

    boolean send(String bindingName, Object data, MimeType outputContentType);

    boolean send(String bindingName, @Nullable String binderName, Object data);

    boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType);
}

通过这几个API就能给消费者发送消息，例如：

1、先声明一个消费者

@Configuration
public class ScsConfiguration {
    private static final Logger log = LoggerFactory.getLogger(ScsConfiguration.class);

    @Bean
    public Consumer<String> consumer() {
        return message -> log.info("我是消费者：接收到消息[bindingName={}] =======> {}", "consumer-in-0", message);
    }
}

2、修改配置文件，定义消费者

spring:
  cloud:
    stream:
      function:
        definition: consumer # 与容器中bean的名称保持一致
      rocketmq:
        binder:
          name-server: 192.168.5.135:9876 # 暂时现在指定namesrv地址(让程序启动不报错)

3、写到这里，先不新增其他东西，我们再来看看新版Spring Cloud Stream到底为我们默认创建了那些东西？

帮我们默认创建了一个名称为consumer-in-0的 binding 并且 consumer-in-0 的 destination 还和名称一致，因此手动发送消息，要想注入容器consumer监听到，bindingName必须写consumer-in-0。

4、生产者手动发送消息

@RestController
@RequestMapping("/rocketmq")
public class SendController {
    private static final Logger log = LoggerFactory.getLogger(SendController.class);

    @Autowired(required = false)
    private StreamBridge streamBridge;

    @GetMapping("/send")
    public R send() {
    	// 发送消息
        boolean send = streamBridge.send("consumer-in-0", "hello rocketmq");
        return R.ok().put("data", send);
    }
}

5、访问API进行测试

6.3、新版本SpringCloudStream集成RocketMQ

了解上面绑定原理后，集成RocketMQ也很简单了。

1、先申明消费者

@Configuration
public class RocketMQConsumerConfiguration {
    private static final Logger log = LoggerFactory.getLogger(RocketMQConsumerConfiguration.class);

    @Autowired(required = false)
    private BindingServiceProperties bindingServiceProperties;

    @Bean
    public Consumer<String> rocketmq() {
        return message -> {
            String methodName = "rocketmq";
            Map<String, BindingProperties> bindings = bindingServiceProperties.getBindings();
            Set<String> bindingNameSet = bindings.keySet();
            Optional<String> bindingOptional = bindingNameSet.stream()
                    .filter(item -> item.startsWith(methodName))
                    .findFirst();
            if (!bindingOptional.isPresent()) {
                return;
            }
            BindingProperties properties = bindings.get(bindingOptional.get());
            log.info("接收到RocketMQ消息[topic={}] =======> {}", properties.getDestination(), message);
            // 处理业务逻辑
        };
    }
}

2、编写配置文件定义生产者和消费者

spring:
  cloud:
    stream:
      function:
        definition: rocketmq
      rocketmq:
        binder:
          name-server: 192.168.5.135:9876
      bindings:
        rocketmq_out: # 生产者 bindingName
          destination: test # topic
          content-type: application/json # 消息类型
          group: rocketmq_out
        rocketmq-in-0: # 消费者 bindingName
          destination: test # topic
          content-type: application/json # 消息类型
          group: rocketmq-in-0
          consumer:
            concurrency: 20
            maxAttempts: 1

3、生产者发送消息

@RestController
@RequestMapping("/rocketmq")
public class SendController {
    @Autowired(required = false)
    private StreamBridge streamBridge;

    @GetMapping("/send")
    public R send() {
        boolean send = streamBridge.send("rocketmq_out", "hello rocketmq");
        return R.ok().put("data", send);
    }
}

4、测试

文章来源地址https://www.toymoban.com/news/detail-601870.html

END

到了这里，关于RocketMQ 5.0 快速入门的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！