SpringBoot整合RocketMQ，三种测试附带源码【rocketmq-spring-boot-starter】

曾经终败给现在 2023-07-12 05:21 23阅读 0赞

> 我们整合boot项目的时候都是引入 xxx-start 依赖，但是现在大多数的整合RocketMQ都还不是这样。  
> 我花了一天时间使用rocketmq-spring-boot-starter整合，使得操作起来更加简单。

## 1、说明 ##

#### 1-1：rocketmq-spring-boot-starter 提供了一个 rocketMQTemplate 使得发消息更加简单，它底层也还是基于DefaultMQProducer 但是封装之后使用更加方便了。 ####

#### 1-2：rocketmq-spring-boot-starter使用监听器的方式获取消息更加的简单 ####

#### 1-3：下面代码提供三种基础的消息发送测试【同步、异步、单向】你只需要修改yml文件里面的RocketMQ地址就好了 ####

#### 1-4：更多内容可以查看官网文档 [https://github.com/apache/rocketmq-spring][https_github.com_apache_rocketmq-spring] ####

## 2、代码 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvbXdpbGRib2Fy_size_16_color_FFFFFF_t_70]

#### 2-1：pom ####

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>www.xdx97.com</groupId>
        <artifactId>RocketMQDemo</artifactId>
        <version>1.0-SNAPSHOT</version>
    
            <dependencies>
                
                <dependency>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot</artifactId>
                    <version>2.2.0.RELEASE</version>
                </dependency>
                
                <dependency>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-web</artifactId>
                    <version>2.2.2.RELEASE</version>
                </dependency>
                
                <dependency>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-spring-boot-starter</artifactId>
                    <version>2.0.2</version>
                </dependency>
                
                <dependency>
                    <groupId>org.apache.rocketmq</groupId>
                    <artifactId>rocketmq-spring-boot-starter</artifactId>
                    <version>2.0.2</version>
                </dependency>
            </dependencies>
    
    </project>

#### 2-1：ProducerController ####

import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    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;
    
    
    /** * 生产者 */
    @RestController
    public class ProducerController { 
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        /** * 同步消息 * * @author 小道仙 * @date 2020年3月3日 */
        @GetMapping("/syncNews")
        public String syncNews(){ 
            Message message = new Message();
            message.setBody("同步消息".getBytes());
            SendResult sendResult = rocketMQTemplate.syncSend("sync-topic", message);
            // 同步消息发送成功会有一个返回值，我们可以用这个返回值进行判断和获取一些信息
            System.out.println(sendResult);
            return "success";
        }
    
        /** * 异步消息 * * @author 小道仙 * @date 2020年3月3日 */
        @GetMapping("/asynNews")
        public String asynNews(){ 
            Message message = new Message();
            message.setBody("异步消息".getBytes());
            rocketMQTemplate.asyncSend("asyn-topic", message, new SendCallback() { 
                public void onSuccess(SendResult sendResult) { 
                    // 成功回调
                }
    
                public void onException(Throwable throwable) { 
                    // 失败回调
                }
            });
            return "success";
        }
    
        /** * 单向消息 * * @author 小道仙 * @date 2020年3月3日 */
        @GetMapping("/sendOneWay")
        public String sendOneWay(){ 
            Message message = new Message();
            message.setBody("单向消息".getBytes());
            rocketMQTemplate.sendOneWay("oneWay-topic",message);
            return "success";
        }
    
    }

#### 2-1：AsynConsumer ####

import org.apache.rocketmq.common.message.MessageExt;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Service;
    
    /** * 异步消息 */
    @Service
    @RocketMQMessageListener(consumerGroup = "my-consumer_asyn-topic", topic = "asyn-topic")
    public class AsynConsumer implements RocketMQListener<MessageExt> { 
    
        public void onMessage(MessageExt messageExt) { 
            byte[] body = messageExt.getBody();
            System.out.println(new String(body));
        }
    }

#### 2-1：OneWayConsumer ####

import org.apache.rocketmq.common.message.MessageExt;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Service;
    
    /** * 单向消息 */
    @Service
    @RocketMQMessageListener(consumerGroup = "my-consumer_oneWay-topic", topic = "oneWay-topic")
    public class OneWayConsumer implements RocketMQListener<MessageExt> { 
    
        public void onMessage(MessageExt messageExt) { 
            byte[] body = messageExt.getBody();
            System.out.println(new String(body));
        }
    }

#### 2-1：SyncConsumer ####

import org.apache.rocketmq.common.message.MessageExt;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Service;
    
    /** * 同步消费 */
    @Service
    @RocketMQMessageListener(consumerGroup = "my-consumer_sync-topic", topic = "sync-topic")
    public class SyncConsumer implements RocketMQListener<MessageExt> { 
    
        public void onMessage(MessageExt messageExt) { 
            byte[] body = messageExt.getBody();
            System.out.println(new String(body));
        }
    }

#### 2-2：RocketMQApp ####

import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class RocketMQApp { 
        public static void main(String[] args) { 
            SpringApplication.run(RocketMQApp.class, args);
        }
    }

#### 2-2：application ####

server:
      port: 80
    
    rocketmq:
      name-server: xxxxxxx:9876
      producer:
        send-message-timeout: 300000
        group: rocketmq-group

## 3、测试 ##

#### 3-1：http://localhost/syncNews ####

#### 3-2：http://localhost/asynNews ####

#### 3-3：http://localhost/sendOneWay ####

ps：我们会看到打印的message信息，你可以使用JSON去格式化获取自己想要的信息。消息有短暂的延迟。

## 4、如果需要源码可以关注微信公众号回复: RocketMQDemo ##

![在这里插入图片描述][20200307130838418.jpg]

[https_github.com_apache_rocketmq-spring]: https://github.com/apache/rocketmq-spring
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvbXdpbGRib2Fy_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200307130859189.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1RvbXdpbGRib2Fy,size_16,color_FFFFFF,t_70
[20200307130838418.jpg]: https://img-blog.csdnimg.cn/20200307130838418.jpg