SpringBoot整合RocketMQ的方法详解
作者：羁客%
这篇文章主要为大家详细介绍了SpringBoot整合RocketMQ的方法，文中的示例代码讲解详细，对我们学习有一定帮助，感兴趣的小伙伴可以了解一下
一：Ubuntu安装RocketMQ
1.下载（在下面地址选择自己需要的版本的rocketmq）
http://rocketmq.apache.org/release_notes/
2.解压，更改配置
将下载的zip文件解压到自己需要安装的位置
在unbuntu系统下需要修改安装跟目录下的两个文件
bin/runserver.sh 和 bin/runbroker.sh 文件
(1).都修改两个文件的 java_home=/java安装HOME路径/
(2).都修改两个文件的 java_home=/java安装HOME路径/jre/
(3).修改runserver.sh内存
JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
# 原来是：JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn4g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
(4).修改runbroker.sh内存
JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g
# 原来是：JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g
(5).创建日志目录
mkdir logs
3.后台挂载启动 mqnamesrv(在安装根目录下执行)
nohup sh bin/mqnamesrv > logs/namesrv.log 2>&1 &
5.查看mqnamesrv是否启动成功
查看 logs/namesrv.log 文件出现以下提示表示启动成功！
The Name Server boot success. serializeType=JSON
6.安装 mqbroker
Broker是 RocketMQ 的核心,大部分'重量级'工作都是由 Broker完成的,
包括接收 Producer 发过来的消息、处理 Consumer 的消费消息请求、消息的持 久化存储、消息的 HA 机制以及服务端过滤功能等 。
(1).更改broker配置，在安装根目录下conf/broker.conf文件 添加一行
brokerIP1 = 127.0.0.0 # IP根据自己电脑服务器修改一下，最好不要使用这个,ifconfig就可以查看到自己服务器的IP
(2).挂载启动 mqbroker（在安装根目录下）
nohup sh bin/mqbroker -c conf/broker.conf -n localhost:9876 > logs/broker.log 2>&1 &
(3).查看 mqbroker 是否启动成功
查看 logs/broker.log 文件出现以下提示表示启动成功！
The broker[broker-a, 127.0.0.0:10911] boot success. serializeType=JSON and name server is localhost:9876
7.关闭broker和nameserver(在安装根目录下执行)
./bin/mqshutdown broker ./bin/mqshutdown namesrv
8.创建一个topic
./bin/mqadmin updateTopic -n '127.0.0.0:9876' -c MyCluster -t MyTopic
9.查看集群状态
./bin/mqadmin clusterList -n localhost:9876
二：添加RocketMQ依赖
<!--rocket mq 依赖-->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.1</version>
<exclusions>
<exclusion>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</exclusion>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
</exclusion>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>spring-webmvc</artifactId>
</exclusion>
</exclusions>
</dependency>
三：在application中添加RocketMQ配置
#rocketmq 配置
rocketmq:
name-server: 127.0.0.0:9876 #IP根据自己电脑服务器配置的 mqbroker->brokerIP1 修改一下，最好不要使用这个
producer:
group: rocketmq_group
myrocketmq-config:
my-topic: rocketmq_topic
my-consumer-group: rocketmq_group_consumer
四：编写消费者，消息生产者，消息实体类(自定义)
//Rocket消费者
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "${myrocketmq-config.my-consumer-group}", topic = "${myrocketmq-config.my-topic}")
public class RocketConsumer implements RocketMQListener<RocketmqVo> {
@Override
public void onMessage(RocketmqVo rocketmqVo) {
log.info("收到RocketMQ消息:{}",rocketmqVo);
}
}
//Rocket生产者
@Slf4j
@Component
public class RocketProduce {
@Value("${myrocketmq-config.my-topic}")
private String mqTopic;
@Value("${myrocketmq-config.my-consumer-group}")
private String mqConsumerGroup;
@Autowired
private RocketMQTemplate mqTemplate;
/**
* 同步发送
*
* @param title 发送消息
* @param data 消息内容
*/
public void sync(String title, Object data) {
String time = new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date());
RocketmqVo rocketmqVo = new RocketmqVo(mqConsumerGroup, mqTopic, title, time, data);
SendResult sendResult = mqTemplate.syncSend(mqTopic, rocketmqVo);
log.info("同步发送:{}", rocketmqVo);
log.info("同步发送消息结果:{}", sendResult);
}
/**
* 异步发送
*
* @param title 发送消息
* @param data 消息内容
*/
public void async(String title, Object data) {
String time = new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date());
RocketmqVo rocketmqVo = new RocketmqVo(mqConsumerGroup, mqTopic, title, time, data);
log.info("异步发送:{}", rocketmqVo);
mqTemplate.asyncSend(mqTopic, rocketmqVo, new SendCallback() {
@Override
public void onSuccess(SendResult var1) {
log.info("异步发送成功:{}", var1);
}
@Override
public void onException(Throwable var1) {
log.info("异步发送失败:{}", var1.getMessage());
}
});
}
/**
* 单向发送
*
* @param title 发送消息
*/
public void oneway(String title, Object data) {
String time = new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date());
RocketmqVo rocketmqVo = new RocketmqVo(mqConsumerGroup, mqTopic, title, time, data);
mqTemplate.sendOneWay(mqTopic, rocketmqVo);
log.info("单向发送:{}", rocketmqVo);
}
}
//Rocket消息实体
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class RocketmqVo {
@Value("${myrocketmq-config.my-consumer-group}")
private String group;
@Value("${myrocketmq-config.my-topic}")
private String topic;
private String title;
private String date;
private Object data;
public RocketmqVo(String title, String date, Object data) {
this.title = title;
this.date = date;
this.data = data;
}
}
五：测试Controller
@RestController
@RequestMapping("/mqtest")
public class RocketController {
@Resource
private RocketProduce producer;
@RequestMapping("/sendMessageSync/{msg}")
public String sendMessageSync(@PathVariable("msg") String message) {
producer.sync(message, null);
return "消息发送完成";
}
@RequestMapping("/sendMessageAsync/{msg}")
public String sendMessageAsync(@PathVariable("msg") String message) {
producer.async(message, null);
return "消息发送完成";
}
@RequestMapping("/sendMessageOneway/{msg}")
public String sendMessageOneway(@PathVariable("msg") String message) {
producer.oneway(message, null);
return "消息发送完成";
}
}以上就是SpringBoot整合RocketMQ的方法详解的详细内容，更多关于SpringBoot整合RocketMQ的资料请关注脚本之家其它相关文章！