当前位置：首页 > news >正文

RocketMQ 详细教程（Spring Boot Spring Cloud Alibaba）

news 来源：原创 2025/5/25 18:55:11

1. RocketMQ 简介

RocketMQ 是阿里巴巴开源的一款分布式消息队列，具有高吞吐量、低延迟、可靠性等特点，广泛应用于金融、电商、物联网等领域。

RocketMQ 的核心特性：
- 高可靠性：支持消息存储、重复消费、失败重试等
- 高可用性：分布式架构，支持主从复制
- 高性能：高吞吐量、低延迟

2. 引入依赖

首先，在 Spring Boot 项目 中引入 RocketMQ 和 Spring Cloud Alibaba 的依赖。

2.1 配置依赖

在 pom.xml 中添加以下依赖：

<dependencies>
    <!-- Spring Boot -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    
    <!-- Spring Cloud Alibaba RocketMQ -->
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-rocketmq</artifactId>
    </dependency>

    <!-- Spring Cloud Alibaba -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba</artifactId>
    </dependency>
</dependencies>

2.2 配置 RocketMQ

在 application.yml 中配置 RocketMQ 连接信息：

spring:
  cloud:
    alibaba:
      rocketmq:
        name-server: 127.0.0.1:9876  # RocketMQ NameServer 地址
        producer:
          group: my-producer-group   # 生产者组
        consumer:
          group: my-consumer-group   # 消费者组

3. 生产者代码实现

3.1 创建消息生产者

我们可以通过注解 @RocketMQMessageListener 来创建 RocketMQ 生产者，消息可以通过 @Value 传递或直接通过 Bean 注入。

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class RocketMQProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 发送普通消息
    public void sendMessage(String topic, String message) {
        rocketMQTemplate.convertAndSend(topic, message);
    }
}

3.2 发送消息

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class ProducerController {

    @Autowired
    private RocketMQProducer rocketMQProducer;

    @GetMapping("/send")
    public String sendMessage() {
        String message = "Hello, RocketMQ!";
        rocketMQProducer.sendMessage("my-topic", message);
        return "Message sent: " + message;
    }
}

访问 http://localhost:8080/send 发送消息，RocketMQ 将开始处理消息。

4. 消费者代码实现

4.1 创建消息消费者

在消费者端，我们需要创建一个消息监听器，利用 @RocketMQMessageListener 注解监听 RocketMQ 消息。

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.stereotype.Service;

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

    @org.springframework.messaging.handler.annotation.MessageMapping
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }
}

4.2 消费消息

每次生产者发送的消息，消费者都会通过 listen 方法进行接收。控制台会打印出收到的消息。

5. 顺序消息

RocketMQ 支持顺序消息，可以通过设置 MessageQueueSelector 来保证消息的顺序性。

5.1 发送顺序消息

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class OrderProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // 发送顺序消息
    public void sendOrderedMessage(String topic, String message, int orderId) {
        rocketMQTemplate.convertAndSend(topic, message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                // 根据orderId选择队列
                return mqs.get(orderId % mqs.size());
            }
        }, orderId);
    }
}

5.2 接收顺序消息

顺序消息的消费逻辑和普通消息相似，只不过要保证顺序消息的消费顺序。

6. 事务消息

RocketMQ 提供了事务消息功能，能够保证消息在分布式事务中的可靠性。我们通过 @RocketMQTransactionListener 来实现。

6.1 配置事务消息

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransaction;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
@RocketMQTransactionListener
public class TransactionProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendTransactionMessage(String topic, String message) {
        rocketMQTemplate.sendMessageInTransaction(topic, message, null);
    }

    public RocketMQLocalTransaction executeLocalTransaction(String message, Object arg) {
        // 事务操作
        try {
            // 业务操作
            return RocketMQLocalTransaction.SUCCESS;
        } catch (Exception e) {
            return RocketMQLocalTransaction.ROLLBACK;
        }
    }
}

6.2 事务回查

RocketMQ 支持事务回查机制，如果事务消息发送后没有明确的提交或回滚，RocketMQ 会通过回查接口查询事务状态。

7. Spring Cloud Alibaba 集成

7.1 配置 Spring Cloud RocketMQ

在 application.yml 配置 Spring Cloud Alibaba 与 RocketMQ 集成。

spring:
  cloud:
    alibaba:
      rocketmq:
        name-server: 127.0.0.1:9876
        producer:
          group: my-producer-group
        consumer:
          group: my-consumer-group

7.2 集成 OpenFeign 与 RocketMQ

通过 OpenFeign 实现远程服务调用，可以和 RocketMQ 一起工作。例如，将 RocketMQ 生产者集成到一个微服务中，使用 OpenFeign 调用。

@FeignClient("rocketmq-producer-service")
public interface RocketMQFeignClient {

    @PostMapping("/send")
    void sendMessage(@RequestBody String message);
}

通过 Feign 客户端发送请求并触发 RocketMQ 生产者的消息发送。

8. 总结

功能总结

功能	说明
消息生产	使用 `RocketMQTemplate` 发送消息
消息消费	使用 `@RocketMQMessageListener` 监听消息
顺序消息	使用 `MessageQueueSelector` 保证消息顺序
事务消息	使用 `RocketMQTransactionListener` 保证事务一致性
集成 Spring Cloud	结合 Spring Cloud Alibaba RocketMQ 进行分布式消息通信