适合新手小白——Spring Boot 整合 RocketMq

爱被打了一巴掌 2023-10-03 21:13 12阅读 0赞

> ![397342b2eaab4927800f661caa44d92b.png][] Apache RocketMQ 自诞生以来，因其架构简单、业务功能丰富、具备极强可扩展性等特点被众多企业开发者以及云厂商广泛采用。历经十余年的大规模场景打磨，RocketMQ 已经成为业内共识的金融级可靠业务消息首选方案，被广泛应用于互联网、大数据、移动互联网、物联网等领域的业务场景。

官方首页：[RocketMQ · 官方网站 | RocketMQ][RocketMQ _ _ _ RocketMQ]

官方文档：[为什么选择RocketMQ | RocketMQ][RocketMQ _ RocketMQ]

下载链接：[https://dlcdn.apache.org/rocketmq/5.0.0/rocketmq-all-5.0.0-bin-release.zip][https_dlcdn.apache.org_rocketmq_5.0.0_rocketmq-all-5.0.0-bin-release.zip]

可视化工具：[GitHub - apache/rocketmq-dashboard: The state-of-the-art Dashboard of Apache RoccketMQ provides excellent monitoring capability. Various graphs and statistics of events, performance and system information of clients and application is evidently made available to the user.][GitHub - apache_rocketmq-dashboard_ The state-of-the-art Dashboard of Apache RoccketMQ provides excellent monitoring capability. Various graphs and statistics of events_ performance and system information of clients and application is evidently made available to the user.]

1.下载安装Apache RocketMQ

![55281ac64a3d4f65946296ae217a9a68.png][]

2.配置系统变量

(1)、ROCKETMQ\_HOME 本地解压路径

![385ae852d55244e3b0fbe6769f6b9901.png][]

(2)、NAMESRV\_ADDR localhost:9876

![7ea3ee0b913c4241a37337bdabe8698a.png][]

3. 启动NameServer

powerShell 管理员身份运行,否则服务会出现闪退现象！！！

若出现以下提示则服务启动成功,可进行下一步操作

![772899a3d0fa4350ae2b39c3411e5d25.png][]

启动报错提示以下错误，请检查服务解压路径中是否包含空格，存在则需迁移修改文件路径

![9f44723009dc4d2db352de0bbfcc8982.png][] 4. 启动Broker+Proxy

单节点则无需分开启动Broker和Proxy，原因官方文档已有说明

![18ba570f17014025a88488020c6898a3.png][]

同样依照步骤三，在服务bin目录下启动mqbroker

![59a8c93bb9cb41ff85c95f2b13d3d23d.png][]5、可视化管理工具——RocketMQ Dashboard

下载zip包,本地解压使用编译工具执行！pathUrl:

![89aa2cd9778c46e0a14fd9dd73c9e2f5.png][]

![9095ac4b4ee141e1980ce7520d7b4a59.png][]

6、测试中间件连通性

访问本地8080端口是否联通正常

![b2932cfcaccd4399bc7be925994bde25.png][]

7、在 pom.xml 文件中添加以下依赖引入Java依赖库

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.2.1</version>
    </dependency>

8、application.properties 配置服务端口及mq连接

server.port=8081
    rocketmq.name-server=127.0.0.1:9876
    rocketmq.producer.group=test1
    rocketmq.producer.send-message-timeout=3000
    rocketmq.producer.compress-message-body-threshold=4096
    rocketmq.producer.max-message-size=4194304
    rocketmq.producer.retry-times-when-send-failed=3
    rocketmq.producer.retry-next-server=true
    rocketmq.producer.retry-times-when-send-async-failed=3

9、创建消息队列![42d1c9d5d259428e9a60103578e8af9d.png][]

10、消息队列生产者，部分推送代码如下：

@Autowired
    private RocketMQTemplate rocketMQTemplate;
    /**
     * 发送普通消息
     * convertAndSend(String destination, Object payload) 发送字符串比较方便
     */
    @RequestMapping("/send")
    public void send(){
        rocketMQTemplate.convertAndSend("test1","test-message");
    }
    /**
     * 发送同步消息
     */
    @RequestMapping("/testSyncSend")
    public void testSyncSend(){
        //参数一：topic   如果想添加tag,可以使用"topic:tag"的写法
        //参数二：消息内容
        SendResult sendResult = rocketMQTemplate.syncSend("test1","同步消息测试");
        System.out.println(sendResult);
    }
    /**
     * 发送异步消息
     */
    @RequestMapping("/testASyncSend")
    public void testASyncSend(){
        //参数一：topic   如果想添加tag,可以使用"topic:tag"的写法
        //参数二：消息内容
        //参数三：回调
        rocketMQTemplate.asyncSend("test1", "异步消息测试", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                System.out.println("消息发送异常");
                throwable.printStackTrace();
            }
        });
    }
    /**
     * 发送单向消息
     */
    @RequestMapping("/testOneWay")
    public void testOneWay(){
        for (int i = 0; i <100000000 ; i++) {
            //参数一：topic   如果想添加tag,可以使用"topic:tag"的写法
            //参数二：消息内容
            rocketMQTemplate.sendOneWay("test1","单向消息测试测试下"+i);
        }
    }

11、消费队列信息消费者

@Service
    @RocketMQMessageListener(consumerGroup = "test1",topic = "test1")
    public class RocketMQConsumerListener implements RocketMQListener<String> {
        @Override
        public void onMessage(String s) {
            System.out.println("消费消息："+s);
        }
    }

12、postmain 测试推送数据并消费 至此完成所有

测试用例已上传至gitee pathUrl：[https://gitee.com/yang-zhaopeng06022021/rocket\_demo.git![icon-default.png?t=M85B][icon-default.png_t_M85B]https://gitee.com/yang-zhaopeng06022021/rocket\_demo.git][https_gitee.com_yang-zhaopeng06022021_rocket_demo.git_icon-default.png_t_M85B_icon-default.png_t_M85B_https_gitee.com_yang-zhaopeng06022021_rocket_demo.git]

[397342b2eaab4927800f661caa44d92b.png]: https://img-blog.csdnimg.cn/397342b2eaab4927800f661caa44d92b.png
[RocketMQ _ _ _ RocketMQ]: https://rocketmq.apache.org/zh/
[RocketMQ _ RocketMQ]: https://rocketmq.apache.org/zh/docs/
[https_dlcdn.apache.org_rocketmq_5.0.0_rocketmq-all-5.0.0-bin-release.zip]: https://dlcdn.apache.org/rocketmq/5.0.0/rocketmq-all-5.0.0-bin-release.zip
[GitHub - apache_rocketmq-dashboard_ The state-of-the-art Dashboard of Apache RoccketMQ provides excellent monitoring capability. Various graphs and statistics of events_ performance and system information of clients and application is evidently made available to the user.]: https://github.com/apache/rocketmq-dashboard
[55281ac64a3d4f65946296ae217a9a68.png]: https://img-blog.csdnimg.cn/55281ac64a3d4f65946296ae217a9a68.png
[385ae852d55244e3b0fbe6769f6b9901.png]: https://img-blog.csdnimg.cn/385ae852d55244e3b0fbe6769f6b9901.png
[7ea3ee0b913c4241a37337bdabe8698a.png]: https://img-blog.csdnimg.cn/7ea3ee0b913c4241a37337bdabe8698a.png
[772899a3d0fa4350ae2b39c3411e5d25.png]: https://img-blog.csdnimg.cn/772899a3d0fa4350ae2b39c3411e5d25.png
[9f44723009dc4d2db352de0bbfcc8982.png]: https://img-blog.csdnimg.cn/9f44723009dc4d2db352de0bbfcc8982.png
[18ba570f17014025a88488020c6898a3.png]: https://img-blog.csdnimg.cn/18ba570f17014025a88488020c6898a3.png
[59a8c93bb9cb41ff85c95f2b13d3d23d.png]: https://img-blog.csdnimg.cn/59a8c93bb9cb41ff85c95f2b13d3d23d.png
[89aa2cd9778c46e0a14fd9dd73c9e2f5.png]: https://img-blog.csdnimg.cn/89aa2cd9778c46e0a14fd9dd73c9e2f5.png
[9095ac4b4ee141e1980ce7520d7b4a59.png]: https://img-blog.csdnimg.cn/9095ac4b4ee141e1980ce7520d7b4a59.png
[b2932cfcaccd4399bc7be925994bde25.png]: https://img-blog.csdnimg.cn/b2932cfcaccd4399bc7be925994bde25.png
[42d1c9d5d259428e9a60103578e8af9d.png]: https://img-blog.csdnimg.cn/42d1c9d5d259428e9a60103578e8af9d.png
[icon-default.png_t_M85B]: https://csdnimg.cn/release/blog_editor_html/release2.2.0/ckeditor/plugins/CsdnLink/icons/icon-default.png?t=M85B
[https_gitee.com_yang-zhaopeng06022021_rocket_demo.git_icon-default.png_t_M85B_icon-default.png_t_M85B_https_gitee.com_yang-zhaopeng06022021_rocket_demo.git]: https://gitee.com/yang-zhaopeng06022021/rocket_demo.git