Spring Boot轻松搞定RocketMQ：消息收发、延迟消息、事务消息与ACL控制

2023-08-22 04:52:43

Spring Boot 与 RocketMQ 集成：消息处理的强大组合

在分布式系统架构的浪潮中，消息队列发挥着举足轻重的作用。RocketMQ 作为一款高性能、高可用、高可靠的开源分布式消息队列，在国内备受推崇。而 Spring Boot 则是一个倍受青睐的 Java 框架，致力于简化微服务应用的构建。

本文将深入探究如何在 Spring Boot 项目中无缝集成 RocketMQ，实现高效的消息发送和接收。

轻松入门

依赖添加

首先，你需要在你的 Spring Boot 项目中引入 RocketMQ 依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-rocketmq</artifactId>
    <version>2.2.5.RELEASE</version>
</dependency>

配置设置

接下来，在配置文件中配置 RocketMQ 相关属性：

rocketmq.name-server=127.0.0.1:9876
rocketmq.producer.group=my-producer-group
rocketmq.consumer.group=my-consumer-group

name-server：指定 RocketMQ 的 Name Server 地址。
producer.group：生产者组名，用于区分不同的消息生产者。
consumer.group：消费者组名，用于区分不同的消息消费者。

发送与接收消息

普通消息

发送消息：

@Service
public class MyProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void send(String message) {
        rocketMQTemplate.convertAndSend("my-topic", message);
    }
}

接收消息：

@RocketMQMessageListener(topic = "my-topic", consumerGroup = "my-consumer-group")
public class MyConsumer {

    @RocketMQMessage(topic = "my-topic")
    public void onMessage(String message) {
        System.out.println("收到消息：" + message);
    }
}

延时消息

发送消息：

public void sendDelay(String message, long delayTime) {
    rocketMQTemplate.convertAndSend("my-topic", message, MessageDelayLevel.ONE_MINUTE);
}

接收消息：

延时消息的接收方式与普通消息类似，消费者会自动根据消息的延迟时间进行处理。

事务消息

发送消息：

@Service
public class MyTransactionProducer {

    @Autowired
    private TransactionMQTemplate transactionMQTemplate;

    public void send(String message) {
        transactionMQTemplate.sendMessageInTransaction("my-topic", message, new ExecuteCallback() {
            @Override
            public TransactionStatus doInTransaction(Message arg0) {
                // 执行本地事务
                return TransactionStatus.COMMIT;
            }
        });
    }
}

接收消息：

事务消息的接收方式与普通消息类似，消费者会自动根据消息的事务状态进行处理。

权限控制

发送端与接收端 ACL

RocketMQ 支持 ACL 控制，允许你通过设置密钥来控制消息的发送和接收权限。

rocketmq.producer.acl-secret-key=my-secret-key
rocketmq.consumer.acl-secret-key=my-secret-key

PULL 消费模式

RocketMQ 支持 PULL 消费模式，允许消费者主动从消息队列中拉取消息。

@ExtRocketMQConsumerConfiguration(
    consumerGroup = "my-consumer-group",
    nameServer = "127.0.0.1:9876",
    topic = "my-topic",
    messageModel = MessageModel.CLUSTERING,
    consumeMode = ConsumeMode.PULL
)
public class MyPullConsumer {

    @RocketMQMessage(topic = "my-topic")
    public void onMessage(String message) {
        System.out.println("收到消息：" + message);
    }
}