spring boot 集成rocketmq

集成Spring Boot和RocketMQ

在现代的微服务架构中，消息队列已经成为一种常见的异步处理模式，它能解决服务间的同步调用、耦合度高、流量高峰等问题。RocketMQ是阿里巴巴开源的一款消息中间件，性能优秀，功能齐全，被广泛应用在各种业务场景。

本文将详细介绍如何在Spring Boot项目中集成RocketMQ，实现消息的生产和消费。

开发环境

• JDK 1.8 或更高
• RocketMQ 4.8.0 或更高
• Spring Boot 2.3.1.RELEASE 或更高
• Maven 3.0 或更高

  RocketMQ服务器部署

  首先，我们需要在本地或服务器上部署RocketMQ。具体的部署步骤可以参考RocketMQ官方文档。为了简化部署，我们可以使用Docker进行部署。

  Spring Boot项目创建

  我们使用Spring Initializr创建一个新的Spring Boot项目，选择Web、Lombok和RocketMQ Spring Boot Starter为项目依赖。

  pom.xml示例：

              org.springframework.boot        spring-boot-starter-web                org.springframework.boot        spring-boot-starter-data-jpa                org.apache.rocketmq        rocketmq-spring-boot-starter        2.1.0                org.projectlombok        lombok    

  配置RocketMQ

  在application.properties文件中配置RocketMQ的服务器地址和其他相关参数。

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

  在这里，rocketmq.name-server是RocketMQ服务器的地址，rocketmq.producer.group是生产者的组名。

  消息生产者

  接下来，我们创建一个消息生产者。在Spring Boot项目中，我们可以使用RocketMQTemplate来发送消息。

  import org.apache.rocketmq.spring.core.RocketMQTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;@RestControllerpublic class ProducerController {    @Autowired    private RocketMQTemplate rocketMQTemplate;    @GetMapping("/send")    public String send(String message) {        rocketMQTemplate.convertAndSend("test-topic", message);        return "Message: '" + message + "' sent.";    }}

  上述代码中，我们创建了一个RESTful接口/send，当接口被调用时，它将发送一个消息到test-topic主题。

  消息消费者

  接下来，我们创建一个消息消费者。在Spring Boot项目中，我们可以使用@RocketMQMessageListener注解来定义一个消息消费者。

  import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;import org.apache.rocketmq.spring.core.RocketMQListener;import org.springframework.stereotype.Service;@Service@RocketMQMessageListener(topic = "test-topic", consumerGroup = "my-consumer_test-topic")public class ConsumerService implements RocketMQListener {    @Override    public void onMessage(String message) {        System.out.printf("------- StringConsumer received: %s \n", message);    }}

  上述代码中，我们定义了一个消息消费者，它将监听test-topic主题的消息，当有新的消息时，它将打印消息内容。

  测试

  至此，我们已经完成了Spring Boot集成RocketMQ的所有代码。接下来，我们就可以运行Spring Boot项目，并通过访问/send接口来发送消息，查看控制台的输出来验证消息消费者是否可以正常接收消息。

  这就是Spring Boot集成RocketMQ的全过程。RocketMQ作为一款功能强大的消息中间件，不仅支持基本的消息生产和消费，还支持许多高级特性，如事务消息、顺序消息、延迟消息等。在实际的项目开发中，我们可以根据业务需求选择合适的消息模型，提高系统的可用性和可靠性。

  事务消息

  RocketMQ支持发送事务消息，也就是说，在发送消息的同时，我们可以执行本地的数据库操作，只有当本地的数据库操作成功时，消息才会真正被发送出去。

  下面是一个发送事务消息的例子：

  import org.apache.rocketmq.client.producer.TransactionListener;import org.apache.rocketmq.client.producer.TransactionMQProducer;import org.apache.rocketmq.spring.core.RocketMQTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;import java.util.concurrent.*;@RestControllerpublic class TransactionProducerController {    @Autowired    private RocketMQTemplate rocketMQTemplate;    @GetMapping("/sendTransaction")    public String sendTransaction(String message) {        ExecutorService executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,                new ArrayBlockingQueue(5000), r -> {            Thread thread = new Thread(r);            thread.setName("client-transaction-msg-check-thread");            return thread;        });        TransactionListener transactionListener = new TransactionListenerImpl();        TransactionMQProducer producer = rocketMQTemplate.createAndStartTransactionMQProducer("transaction-group",transactionListener,executor);        producer.sendMessageInTransaction("test-topic", "TagA", message, null);        return "Transaction Message: '" + message + "' sent.";    }}

  在上述代码中，我们创建了一个TransactionMQProducer，并设置了一个TransactionListener来处理事务的提交和回滚。当发送事务消息时，我们需要调用sendMessageInTransaction方法。

  顺序消息

  RocketMQ支持发送顺序消息，也就是说，消息会按照发送的顺序被消费。

  下面是一个发送顺序消息的例子：

  import org.apache.rocketmq.spring.core.RocketMQTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;import org.springframework.messaging.support.MessageBuilder;@RestControllerpublic class OrderlyProducerController {    @Autowired    private RocketMQTemplate rocketMQTemplate;    @GetMapping("/sendOrderly")    public String sendOrderly(String message) {        for (int i = 0; i  
  在上述代码中，我们调用syncSendOrderly方法发送顺序消息。该方法的第三个参数是hashkey，RocketMQ会根据这个key来决定消息发送到哪个队列，具有相同hashkey的消息会发送到同一个队列。
   
  延迟消息
   
  RocketMQ支持发送延迟消息，也就是说，消息不会立即被消费，而是会在指定的时间后被消费。
   
  下面是一个发送延迟消息的例子：
   
  import org.apache.rocketmq.spring.core.RocketMQTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RestController;import org.springframework.messaging.support.MessageBuilder;@RestControllerpublic class DelayProducerController {    @Autowired    private RocketMQTemplate rocketMQTemplate;    @GetMapping("/sendDelay")    public String sendDelay(String message) {        rocketMQTemplate.syncSend("delay_topic", MessageBuilder.withPayload(message).build(), 1000, 4);        return "Delay Message: '" + message + "' sent.";    }}
   
  在上述代码中，我们调用syncSend方法发送延迟消息。该方法的第三个参数是延迟时间，第四个参数是延迟级别。