脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → SpringBoot整合RocketMQ

SpringBoot整合RocketMQ的方法详解

 更新时间：2022年08月12日 14:38:39   作者：羁客%  
这篇文章主要为大家详细介绍了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的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

相关文章

  • Java数据结构中堆的向下和向上调整解析

    Java数据结构中堆的向下和向上调整解析

    堆是一颗完全二叉树，在这棵树中，所有父节点都满足大于等于其子节点的堆叫大根堆，所有父节点都满足小于等于其子节点的堆叫小根堆。堆虽然是一颗树，但是通常存放在一个数组中，父节点和孩子节点的父子关系通过数组下标来确定
    2021-11-11
  • java仿枚举实例

    java仿枚举实例

    下面小编就为大家带来一篇java仿枚举实例。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
    2017-08-08
  • MyBatis映射器mapper快速入门教程

    MyBatis映射器mapper快速入门教程

    使用mapper接口的方式，不用写接口实现类，直接完成数据库操作，简单方便。使用mapper接口，采用的是面向接口编程的思想，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习吧
    2022-09-09
  • springboot快速搭建ftpserver服务端的详细步骤

    springboot快速搭建ftpserver服务端的详细步骤

    基于springboot,使用ftpserver快速搭建一个FTP服务端,搭建过程很简单,我们把过程分成4个步骤,一分钟内快速完成构建,感兴趣的朋友跟随小编一起看看吧
    2023-11-11
  • Spring IOC简单理解及创建对象的方式

    Spring IOC简单理解及创建对象的方式

    这篇文章主要介绍了Spring IOC简单理解及创建对象的方式,本文通过两种方式给大家介绍创建对象的方法，通过实例代码给大家介绍的非常详细，需要的朋友可以参考下
    2021-09-09
  • 一篇文章带你复习java知识点

    一篇文章带你复习java知识点

    以下简单介绍了下我对于这些java基本知识点和技术点的一些看法和心得，这些内容都源自于我这些年来使用java的一些总结,希望能够给你带来帮助
    2021-06-06
  • Java实现批量合并Excel工作表

    Java实现批量合并Excel工作表

    这篇文章主要为大家详细介绍了如何使用Java快速实现批量Excel工作表的合并,文中的示例代码代码讲解详细,有需要的小伙伴可以跟随小编一起学习一下
    2023-10-10
  • Springboot通过lucene实现全文检索详解流程

    Springboot通过lucene实现全文检索详解流程

    Lucene是一个基于Java的全文信息检索工具包，它不是一个完整的搜索应用程序，而是为你的应用程序提供索引和搜索功能。Lucene 目前是 Apache Jakarta 家族中的一个开源项目，也是目前最为流行的基于 Java 开源全文检索工具包
    2022-06-06
  • Spring Boot整合Spring Cache及Redis过程解析

    Spring Boot整合Spring Cache及Redis过程解析

    这篇文章主要介绍了Spring Boot整合Spring Cache及Redis过程解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2019-12-12
  • SpringBoot加载静态资源的方式

    SpringBoot加载静态资源的方式

    本篇文章主要介绍了SpringBoot加载静态资源的方式，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
    2017-04-04

最新评论