springBoot整合RocketMQ及坑的示例代码

接下来我将详细讲解“springBoot整合RocketMQ及坑的示例代码”的完整攻略。

一、背景

在我们使用Spring Boot构建分布式系统时，经常会用到消息队列。RocketMQ是阿里巴巴的开源消息中间件，它支持事务消息，并且具有高吞吐量、高可用性、高可靠性等特点，十分适合在分布式系统中使用。本文将介绍如何在Spring Boot项目中整合RocketMQ，并附带两条示例说明。

二、整合步骤

2.1 引入依赖

首先需要在pom.xml文件中引入RocketMQ的依赖：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>${rocketmq.version}</version>
</dependency>

其中${rocketmq.version}是RocketMQ版本号变量，需要在<properties>标签中定义。

2.2 配置RocketMQ

在application.yml或application.properties中添加RocketMQ的配置项，如下所示：

rocketmq:
  name-server: localhost:9876

其中name-server为RocketMQ的nameserver地址，默认为localhost:9876。

2.3 创建生产者和消费者

在Spring Boot项目中，我们可以使用@Component或@Service注解定义RocketMQ的生产者和消费者。以下是两个示例：

2.3.1 生产者示例

@Component
public class RocketMQProducer {

    @Autowired
    private DefaultMQProducer producer;

    public void send(String topic, String message) throws Exception {
        Message msg = new Message(topic, message.getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult result = producer.send(msg);
        System.out.printf("Send message success. Result: %s%n", result);
    }
}

在上述示例中，我们使用@Component注解将一个名为RocketMQProducer的组件注入到Spring容器中。该组件使用@Autowired注解自动装配RocketMQ的默认生产者DefaultMQProducer。send方法用于发送消息，其中msg变量为RocketMQ的消息实体，SendResult变量为消息发送结果。

2.3.2 消费者示例

@Service
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "test-group")
public class RocketMQConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.printf("Received message: %s%n", message);
    }
}

在上述示例中，我们使用@Service注解将一个名为RocketMQConsumer的组件注入到Spring容器中。该组件实现了RocketMQ的监听器接口RocketMQListener，用于处理接收到的消息。@RocketMQMessageListener注解指定了需要监听的消息主题test-topic以及消费者组test-group。

2.4 配置生产者和消费者

最后，我们需要在Spring Boot项目的配置类中配置生产者和消费者，如下所示：

@Configuration
public class RocketMQConfig {

    @Value("${rocketmq.name-server}")
    private String namesrvAddr;

    @Bean
    public DefaultMQProducer defaultMQProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("default-group");
        producer.setNamesrvAddr(namesrvAddr);
        producer.start();
        return producer;
    }
}

在上述示例中，我们使用@Configuration注解将一个名为RocketMQConfig的配置类注入到Spring容器中。@Value注解可以将application.yml或application.properties中的属性值注入到变量中。defaultMQProducer方法通过@Bean注解将一个名为defaultMQProducer的生产者注入到Spring容器中，该生产者的组名为default-group，连接的nameserver地址为namesrvAddr。

至此，我们已经完成了Spring Boot整合RocketMQ的步骤。

三、示例说明

下面分别通过点对点和发布订阅两种模式，说明如何使用上述的RocketMQ生产者和消费者。

3.1 点对点示例

在点对点模式中，一条消息只会被一个消费者接收。以下是点对点模式的示例代码：

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

    @Autowired
    private RocketMQProducer producer;

    @PostMapping("/send")
    public ResponseEntity<String> send(String message) {
        try {
            producer.send("test-topic", message);
            return ResponseEntity.ok("Send message success.");
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("Send message failed.");
        }
    }
}

在上述示例中，我们使用@RestController和@RequestMapping注解定义了一个HTTP接口/rocketmq/send，该接口可以通过POST请求发送消息。当请求到达后，我们使用producer对象发送消息，其中"test-topic"是消息主题，message是消息内容。

在前面定义的RocketMQConsumer中，我们需要新增selectorExpression属性指定消息的子表达式。修改后的示例如下：

@Service
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "test-group", selectorExpression = "test-tag")
public class RocketMQConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.printf("Received message: %s%n", message);
    }
}

在上述示例中，我们新增了selectorExpression属性，指定消息的子表达式为test-tag。这样，该消费者只会接收到主题为test-topic且标签为test-tag的消息。

3.2 发布订阅示例

在发布订阅模式中，一条消息可以被多个消费者接收。以下是发布订阅模式的示例代码：

@Service
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "test-group")
public class RocketMQSubscriber implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.printf("Received message: %s%n", message);
    }
}

在上述示例中，我们使用@Service注解定义了一个名为RocketMQSubscriber的组件，该组件实现了RocketMQ的监听器接口RocketMQListener。在@RocketMQMessageListener注解中指定了需要监听的消息主题test-topic以及消费者组test-group，可以同时启动多个消费者实例进行消息消费。

在生产者示例中，修改send方法，增加消息标签tags。示例如下：

@Component
public class RocketMQProducer {

    @Autowired
    private DefaultMQProducer producer;

    public void send(String topic, String message) throws Exception {
        Message msg = new Message(topic, "test-tag", message.getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult result = producer.send(msg);
        System.out.printf("Send message success. Result: %s%n", result);
    }
}

在上述示例中，我们在Message构造函数中新增了"test-tag"作为消息标签。这样，发布订阅模式中，消费者监听到主题为test-topic的消息时，只有标签为test-tag的消息才会被接收。

四、总结

本文介绍了如何在Spring Boot项目中整合RocketMQ，并提供了点对点和发布订阅两种模式的示例代码。在使用RocketMQ时，需要注意一些坑点，例如在消费者中新增selectorExpression属性指定消息的子表达式，或者在发送时同时指定消息主题和标签等。掌握了这些技巧后，我们可以更加灵活地应用RocketMQ，并快速构建高可靠性分布式系统。