个人中心
gpt4 book ai didi

springBoot整合RocketMQ及坑的示例代码

转载 作者：qq735679552 更新时间：2022-09-28 22:32:09 26 4
gpt4 key购买 nike

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章springBoot整合RocketMQ及坑的示例代码由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

版本:

  • jdk:1.8
  • springboot:1.5.10
  • rocketmq：4.2.0

pom 配置：     。

?
1
2
3
4
5
6
7
8
9
10
<parent>
  <groupid>org.springframework.boot</groupid>
  <artifactid>spring-boot-starter-parent</artifactid>
  <version> 1.5 . 10 .release</version>
</parent>
<dependency>
   <groupid>org.apache.rocketmq</groupid>
   <artifactid>rocketmq-client</artifactid>
   <version> 4.2 . 0 </version>
</dependency>

application.properties  配置

?
1
2
3
4
5
6
# 消费者的组名
apache.rocketmq.consumer.pushconsumer=pushconsumer
# 生产者的组名
apache.rocketmq.producer.producergroup=producer
# nameserver地址
apache.rocketmq.namesrvaddr=localhost: 9876

java代码:

生产者 。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
package test.config.rocketmq;
 
import org.apache.rocketmq.client.producer.defaultmqproducer;
import org.apache.rocketmq.client.producer.sendresult;
import org.apache.rocketmq.common.message.message;
import org.apache.rocketmq.remoting.common.remotinghelper;
import org.springframework.beans.factory.annotation.value;
import org.springframework.stereotype.component;
import org.springframework.util.stopwatch;
import javax.annotation.postconstruct;
 
@component
public class rocketmqclient {
   /**
    * 生产者的组名
    */
   @value ( "${apache.rocketmq.producer.producergroup}" )
   private string producergroup;
 
   /**
    * nameserver 地址
    */
   @value ( "${apache.rocketmq.namesrvaddr}" )
   private string namesrvaddr;
 
   @postconstruct
   public void defaultmqproducer() {
     //生产者的组名
     defaultmqproducer producer = new defaultmqproducer(producergroup);
     //指定nameserver地址，多个地址以 ; 隔开
     producer.setnamesrvaddr(namesrvaddr);
     producer.setvipchannelenabled( false );
     try {
       /**
        * producer对象在使用之前必须要调用start初始化，初始化一次即可
        * 注意：切记不可以在每次发送消息时，都调用start方法
        */
       producer.start();
 
       //创建一个消息实例，包含 topic、tag 和 消息体
       //如下：topic 为 "topictest"，tag 为 "push"
       message message = new message( "topictest" , "push" , "发送消息----zhisheng-----" .getbytes(remotinghelper.default_charset));
 
       stopwatch stop = new stopwatch();
       stop.start();
 
       for ( int i = 0 ; i < 1 ; i++) {
         sendresult result = producer.send(message);
         system.out.println( "发送响应：msgid:" + result.getmsgid() + "，发送状态:" + result.getsendstatus());
       }
       stop.stop();
       system.out.println( "----------------发送一万条消息耗时：" + stop.gettotaltimemillis());
     } catch (exception e) {
       e.printstacktrace();
     } finally {
       producer.shutdown();
     }
   }
}

消费者：  。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import org.apache.rocketmq.client.consumer.defaultmqpushconsumer;
import org.apache.rocketmq.client.consumer.listener.consumeconcurrentlystatus;
import org.apache.rocketmq.client.consumer.listener.messagelistenerconcurrently;
import org.apache.rocketmq.common.consumer.consumefromwhere;
import org.apache.rocketmq.common.message.messageext;
import org.apache.rocketmq.remoting.common.remotinghelper;
import org.springframework.beans.factory.annotation.value;
import org.springframework.stereotype.component;
 
import javax.annotation.postconstruct;
 
 
@component
public class rocketmqserver {
   /**
    * 消费者的组名
    */
   @value ( "${apache.rocketmq.consumer.pushconsumer}" )
   private string consumergroup;
 
   /**
    * nameserver 地址
    */
   @value ( "${apache.rocketmq.namesrvaddr}" )
   private string namesrvaddr;
 
   @postconstruct
   public void defaultmqpushconsumer() {
     //消费者的组名
     defaultmqpushconsumer consumer = new defaultmqpushconsumer(consumergroup);
 
     //指定nameserver地址，多个地址以 ; 隔开
     consumer.setnamesrvaddr(namesrvaddr);
     consumer.setvipchannelenabled( false );
     try {
       //订阅pushtopic下tag为push的消息
       consumer.subscribe( "topictest" , "push" );
 
       //设置consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
       //如果非第一次启动，那么按照上次消费的位置继续消费
       consumer.setconsumefromwhere(consumefromwhere.consume_from_first_offset);
       consumer.registermessagelistener((messagelistenerconcurrently) (list, context) -> {
         try {
           for (messageext messageext : list) {
 
             system.out.println( "messageext: " + messageext); //输出消息内容
 
             string messagebody = new string(messageext.getbody(), remotinghelper.default_charset);
 
             system.out.println( "消费响应：msgid : " + messageext.getmsgid() + ", msgbody : " + messagebody); //输出消息内容
           }
         } catch (exception e) {
           e.printstacktrace();
           return consumeconcurrentlystatus.reconsume_later; //稍后再试
         }
         return consumeconcurrentlystatus.consume_success; //消费成功
       });
       consumer.start();
     } catch (exception e) {
       e.printstacktrace();
     }
   }
}

掉坑总结:

1.rocketmq启动时，命令不是  mqbroker -n 127.0.0.1:9876 。

         正确应该是：mqbroker -n 127.0.0.1:9876 butiautocreatetopicenable=true 。

         否则会抛出：no route info of this topic, topictest 。

2.客户端连接时抛出异常 。

        org.apache.rocketmq.client.exception.mqclientexception:  。

        send [3] times, still failed, cost [3180]ms, topic: topictest, brokerssent: \ 。

        [win-93cgo0s5g25, win-93cgo0s5g25, win-93cgo0s5g25] 。

解决方式两种 。

1.producer.setvipchannelenabled(false); 生产者和消费者添加这行代买.

2.降rocketmq版本,降成3.2.6 。

关于spring.rocketmq.name-server的坑 。

看下图:

springBoot整合RocketMQ及坑的示例代码

注意:

如果你是springboot2.0+的框架，或者是jdk10.

你需要将你自己的项目配置文件中的，spring.rocketmq.name-server改成 。

spring.rocketmq.nameserver。注意是nameserver.

不然就会报各种稀奇古怪的bug.

关于启动报内存不足的错 。

在安装启动name server和broker的时候，一定要修改配置文件，不然内存会爆炸.

native memory allocation (mmap) failed to map 8589934592 bytes for committing reserved memory  。

springBoot整合RocketMQ及坑的示例代码

将下面的配置文件根据你的需要改 。

我这里以前默认是xms4g，都是g，我修改到m就行了.

java_opt="${java_opt} -server -xms256m -xmx256m -xmn128m -xx:metaspacesize=128m -xx:maxmetaspacesize=320m" 。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持我.

原文链接：https://blog.csdn.net/qq_24853627/article/details/79443437 。

最后此篇关于springBoot整合RocketMQ及坑的示例代码的文章就讲到这里了,如果你想了解更多关于springBoot整合RocketMQ及坑的示例代码的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。

26 4 0
文章推荐： Java8的常用时间api实用指南
文章推荐： C语言实现贪吃蛇小游戏
文章推荐： C语言实现拼图小游戏
文章推荐： Spring Boot系列教程之日志配置
qq735679552
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com