SpringBoot整合RocketMQ实现入门案例

Posted 2022-04-05 刘Java

此前我们学习了如何在linux安装RocketMQ，以及使用Java程序公网连接的一系列准备工作，并且学习了Spring整合RocketMQ的第一个案例！

现在我们来学习Spring Boot如何整合RocketMQ实现更加简单的使用！

文章目录

• 1 创建maven项目
• 2 配置文件
• 3 生产者
• 4 消费者
• 5 测试

1 创建maven项目

创建一个maven项目。引入springboot和rocketmq的pom依赖。

为了让用户能够更灵活、方便地结合Spring Boot使用RocketMQ消息客户端，Apache RocketMQ社区推出了spring-boot-starter实现，因此这里我们不需要引入原始的rocketmq-client，而是引入rocketmq-spring-boot-starter。

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.3.5.RELEASE</version>
    <relativePath/>
</parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <!--RocketMQ依赖-->
    <!-- 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.1</version>
    </dependency>
</dependencies>

建立如下目录结构：

项目启动类RocketMQApplication：

@SpringBootApplication
public class RocketmqSpringbootApplication

   public static void main( String[] args )

      SpringApplication.run( RocketmqSpringbootApplication.class, args );
   

2 配置文件

在配置文件中，我们编写rocketmq的配置信息。需要注意的是，如果你配置了rocketmq:consumer下面的参数，那么将会默认创建一个consumer。

rocketmq:
  producer:
    group: bootProducer
  name-server: xx.94.xx.xx:9876
  #自定义的配置
  topic: bootTopic
  group: bootConsumer

3 生产者

通过引入RocketMQTemplate代替DefaultMQProducer对象作为消息生产者，RocketMQTemplate被Spring boot自动配置，无需手动配置，我们直接引入即可。

@RestController
public class MqController

   /**
    * 引入template代替defaultMQProducer作为生产者
    */
   @Resource
   private RocketMQTemplate rocketMQTemplate;

   @Value( "$rocketmq.topic" )
   private String topic;

   @GetMapping( "/testMq/flag" )
   public String testMq( @PathVariable @NotNull String flag )
      //通过MessageBuilder构建消息
      Message<String> message = MessageBuilder.withPayload( flag ).build();
      //同步发送该消息，获取发送结果
      SendResult result = rocketMQTemplate.syncSend( topic, message );
      return result.getMsgId() + "----" + result.getSendStatus();
   

4 消费者

通过实现RocketMQListener接口的对象作为消费者，实现onMessage方法，该方法的参数就是消息内容，通过在类上配置@RocketMQMessageListener注解将其包装为一个RocketMQ的消费者并自动启动。

@Service
@RocketMQMessageListener( consumerGroup = "$rocketmq.group", topic = "$rocketmq.topic" )
public class Consumer implements RocketMQListener<String>

   private static final Logger logger = LoggerFactory.getLogger( Consumer.class );

   @Override
   public void onMessage( String message )

      System.out.println( "message:  " + message );
      //模拟消费失败，抛出异常
      if( "1".equals( message ) )
         message = null;
         message.toString();
      
   

5 测试

启动项目。访问http://localhost:8080/testMq/xxx发送消息，可以观察到消息消费成功，访问http://localhost:8080/testMq/1发送错误消息，可以观察到消息消费失败，并且重试消费。关于消息重试：RocketMQ的消息重试（消息重投）。

不建议使用RocketMQTemplate或者@RocketMQMessageListener的方式收发消息，因为他们提供的自动配置属性并不全，并且如果不熟悉自动配置的原理的话可能会有个好走不过你的坑，建议使用此前的配置。