RocketMQ保姆级教程

news2024/9/30 9:19:47

RocketMQ是阿里巴巴旗下一款开源的MQ框架，经历过双十一考验、Java编程语言实现，有非常好完整生态系统。RocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等，总之就是葛大爷的一句话

图片

整篇文章可以大致分为三个部分，第一部分属于一些核心概念和工作流程的讲解；第二部分就是纯手动搭建了一套环境；第三部分是基于环境进行测试和集成到SpringBoot，因为整个过程讲的比较细，所以我称之为“保姆级教程”。

好了，废话补多少，直接进入主题。

核心概念

  • NameServer：可以理解为是一个注册中心，主要是用来保存topic路由信息，管理Broker。在NameServer的集群中，NameServer与NameServer之间是没有任何通信的。

  • Broker：核心的一个角色，主要是用来保存topic的信息，接受生产者产生的消息，持久化消息。在一个Broker集群中，相同的BrokerName可以称为一个Broker组，一个Broker组中,BrokerId为0的为主节点，其它的为从节点。BrokerName和BrokerId是可以在Broker启动时通过配置文件配置的。每个Broker组只存放一部分消息。

  • 生产者：生产消息的一方就是生产者

  • 生产者组：一个生产者组可以有很多生产者，只需要在创建生产者的时候指定生产者组，那么这个生产者就在那个生产者组

  • 消费者：用来消费生产者消息的一方

  • 消费者组：跟生产者一样，每个消费者都有所在的消费者组，一个消费者组可以有很多的消费者，不同的消费者组消费消息是互不影响的。

  • topic（主题）：可以理解为一个消息的集合的名字，生产者在发送消息的时候需要指定发到哪个topic下，消费者消费消息的时候也需要知道自己消费的是哪些topic底下的消息。

  • Tag（子主题）：比topic低一级，可以用来区分同一topic下的不同业务类型的消息，发送消息的时候也需要指定。

这里有组的概念是因为可以用来做到不同的生产者组或者消费者组有不同的配置，这样就可以使得生产者或者消费者更加灵活。

工作流程

说完核心概念，再来说一下核心的工作流程，这里我先画了一张图。

图片

通过这张图就可以很清楚的知道，RocketMQ大致的工作流程：

  • Broker启动的时候，会往每台NameServer（因为NameServer之间不通信，所以每台都得注册）注册自己的信息，这些信息包括自己的ip和端口号，自己这台Broker有哪些topic等信息。

  • Producer在启动之后会跟会NameServer建立连接，定期从NameServer中获取Broker的信息，当发送消息的时候，会根据消息需要发送到哪个topic去找对应的Broker地址，如果有的话，就向这台Broker发送请求；没有找到的话，就看根据是否允许自动创建topic来决定是否发送消息。

  • Broker在接收到Producer的消息之后，会将消息存起来，持久化，如果有从节点的话，也会主动同步给从节点，实现数据的备份

  • Consumer启动之后也会跟会NameServer建立连接，定期从NameServer中获取Broker和对应topic的信息，然后根据自己需要订阅的topic信息找到对应的Broker的地址，然后跟Broker建立连接，获取消息，进行消费

就跟上面的图一样，整体的工作流程还是比较简单的，这里我简化了很多概念，主要是为了好理解。

环境搭建

终于讲完了一些简单的概念，接下来就来搭建一套RocketMQ的环境。

通过上面分析，我们知道，在RocketMQ中有NameServer、Broker、生产者、消费者四种角色。而生产者和消费者实际上就是业务系统，所以这里不需要搭建，真正要搭建的就是NameServer和Broker，但是为了方便RocketMQ数据的可视化，这里我多搭建一套可视化的服务。

搭建过程比较简单，按照步骤一步一步来就可以完成，如果提示一些命令不存在，那么直接通过yum安装这些命令就行。

一、准备

需要准备一个linux服务器，需要先安装好JDK

关闭防火墙

systemctl stop firewalld
systemctl disable firewalld
下载并解压RocketMQ
1、创建一个目录，用来存放rocketmq相关的东西
mkdir /usr/rocketmq
cd /usr/rocketmq
2、下载并解压rocketmq

下载

wget https://archive.apache.org/dist/rocketmq/4.7.1/rocketmq-all-4.7.1-bin-release.zip

解压

unzip rocketmq-all-4.7.1-bin-release.zip

看到这一个文件夹就完成了

图片

然后进入rocketmq-all-4.7.1-bin-release文件夹

cd rocketmq-all-4.7.1-bin-release

RocketMQ的东西都在这了

图片

二、搭建NameServer

修改jvm参数

在启动NameServer之前，强烈建议修改一下启动时的jvm参数，因为默认的参数都比较大，为了避免内存不够，建议修改小，当然，如果你的内存足够大，可以忽略。

vi bin/runserver.sh

修改画圈的这一行

图片

这里你可以直接修改成跟我一样的

-server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=32m -XX:MaxMetaspaceSize=50m
启动NameServer

修改完之后，执行如下命令就可以启动NameServer了

nohup sh bin/mqnamesrv &

查看NameServer日志

tail -f ~/logs/rocketmqlogs/namesrv.log

如果看到如下的日志，就说明启动成功了

图片

NameServer日志

三、搭建Broker

这里启动单机版的Broker

修改jvm参数

跟启动NameServer一样，也建议去修改jvm参数

vi bin/runbroker.sh

将画圈的地方设置小点，当然也别太小啊

图片

当然你还是可以跟我设置的一样

-server -Xms1g -Xmx1g -Xmn512m
修改Broker配置文件broker.conf

这里需要改一下Broker配置文件，需要指定NameServer的地址，因为需要Broker需要往NameServer注册

vi conf/broker.conf

Broker配置文件

图片

Broker配置文件

这里就能看出Broker的配置了，什么Broker集群的名称啊，Broker的名称啊，Broker的id啊，都跟前面说的对上了。

在文件末尾追加地址

namesrvAddr = localhost:9876

因为NameServer跟Broker在同一台机器，所以是localhost，NameServer端口默认的是9876。

不过这里我还建议再修改一处信息，因为Broker向NameServer进行注册的时候，带过去的ip如果不指定就会自动获取，但是自动获取的有个坑，就是有可能你的电脑无法访问到这个自动获取的ip，所以我建议手动指定你的电脑可以访问到的服务器ip。

我的虚拟机的ip是192.168.200.143，所以就指定为192.168.200.143，如下

brokerIP1 = 192.168.200.143
brokerIP2 = 192.168.200.143

如果以上都配置的话，最终的配置文件应该如下，红圈的为新加的

图片

启动Broker
nohup sh bin/mqbroker -c conf/broker.conf &

-c 参数就是指定配置文件

查看日志

tail -f ~/logs/rocketmqlogs/broker.log

当看到如下日志就说明启动成功了

图片

四、搭建可视化控制台

其实前面NameServer和Broker搭建完成之后，就可以用来收发消息了，但是为了更加直观，可以搭一套可视化的服务。

可视化服务其实就是一个jar包，启动就行了。

jar包可以从这获取

链接：https://pan.baidu.com/s/16s1qwY2qzE2bxR81t5Wm6w
提取码：s0sd

将jar包上传到服务器，放到/usr/rocketmq的目录底下，当然放哪都无所谓，这里只是为了方便，因为rocketmq的东西都在这里

然后进入/usr/rocketmq下，执行如下命名

nohup java -jar -server -Xms256m -Xmx256m -Drocketmq.config.namesrvAddr=localhost:9876 -Dserver.port=8088 rocketmq-console-ng-1.0.1.jar &

rocketmq.config.namesrvAddr就是用来指定NameServer的地址的

查看日志

tail -f ~/logs/consolelogs/rocketmq-console.log

当看到如下日志，就说明启动成功了

图片

然后在浏览器中输入http://linux服务器的ip:8088/就可以看到控制台了，如果无法访问，可以看看防火墙有没有关闭

图片

右上角可以把语言切换成中文

图片

Broker集群信息

图片

topic信息

通过控制台可以查看生产者、消费者、Broker集群等信息，非常直观。

功能很多，这里就不一一介绍了。

测试

环境搭好之后，就可以进行测试了。

引入依赖
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.7.1</version>
</dependency>
生产者发送消息
public class Producer {
    public static void main(String[] args) throws Exception {
        //创建一个生产者，指定生产者组为sanyouProducer
        DefaultMQProducer producer = new DefaultMQProducer("sanyouProducer");

        // 指定NameServer的地址
        producer.setNamesrvAddr("192.168.200.143:9876");
        // 第一次发送可能会超时，我设置的比较大
        producer.setSendMsgTimeout(60000);

        // 启动生产者
        producer.start();

        // 创建一条消息
        // topic为 sanyouTopic
        // 消息内容为 三友的java日记
        // tags 为 TagA
        Message msg = new Message("sanyouTopic", "TagA", "三友的java日记 ".getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 发送消息并得到消息的发送结果，然后打印
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);

        // 关闭生产者
        producer.shutdown();
    }

}

  • 构建一个消息生产者DefaultMQProducer实例，然后指定生产者组为sanyouProducer；

  • 指定NameServer的地址：服务器的ip:9876，因为需要从NameServer拉取Broker的信息

  • producer.start() 启动生产者

  • 构建一个内容为三友的java日记的消息，然后指定这个消息往sanyouTopic这个topic发送

  • producer.send(msg)：发送消息，打印结果

  • 关闭生产者

运行结果如下

SendResult [sendStatus=SEND_OK, msgId=C0A81FAF54F818B4AAC2475FD2010000, offsetMsgId=C0A8C88F00002A9F000000000009AE55, messageQueue=MessageQueue [topic=sanyouTopic, brokerName=broker-a, queueId=0], queueOffset=0]

sendStatus=SEND_OK 说明发送成功了，此时就能后控制台看到未消费的消息了。

图片

到控制台看到消息那块，然后选定发送的topic，查询的时间范围手动再选一下，不选就查不出来(我怀疑这是个bug)，然后查询就能看到了一条消息。

然后点击一下MESSAGE DETAIL就能够看到详情。

图片

这里就能看到发送消息的详细信息。

左下角消息的消费的消费，因为我们还没有消费者订阅这个topic，所以左下角没数据。

消费者消费消息
public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {

        // 通过push模式消费消息，指定消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sanyouConsumer");

        // 指定NameServer的地址
        consumer.setNamesrvAddr("192.168.200.143:9876");

        // 订阅这个topic下的所有的消息
        consumer.subscribe("sanyouTopic", "*");

        // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("消费消息:%s", new String(msg.getBody()) + "\n");
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}

  • 创建一个消费者实例对象，指定消费者组为sanyouConsumer

  • 指定NameServer的地址：服务器的ip:9876

  • 订阅 sanyouTopic 这个topic的所有信息

  • consumer.registerMessageListener ，这个很重要，是注册一个监听器，这个监听器是当有消息的时候就会回调这个监听器，处理消息，所以需要用户实现这个接口，然后处理消息。

  • 启动消费者

启动之后，消费者就会消费刚才生产者发送的消息，于是控制台就打印出如下信息

Consumer Started.
消费消息:三友的java日记 

此时再去看控制台

图片

发现被sanyouConsumer这个消费者组给消费了。

SpringBoot环境下集成RocketMQ

集成

在实际项目中肯定不会像上面测试那样用，都是集成SpringBoot的。

1、引入依赖
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.1</version>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <version>2.1.1.RELEASE</version>
</dependency>
2、yml配置
rocketmq:
  producer:
    group: sanyouProducer
  name-server: 192.168.200.143:9876
3、创建消费者

SpringBoot底下只需要实现RocketMQListener接口，然后加上@RocketMQMessageListener注解即可

@Component
@RocketMQMessageListener(consumerGroup = "sanyouConsumer", topic = "sanyouTopic")
public class SanYouTopicListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        System.out.println("处理消息:" + msg);
    }

}

@RocketMQMessageListener需要指定消费者属于哪个消费者组，消费哪个topic，NameServer的地址已经通过yml配置文件配置类

4、测试
@SpringBootTest(classes = RocketMQApplication.class)
@RunWith(SpringRunner.class)
public class RocketMQTest {

    @Autowired
    private RocketMQTemplate template;

    @Test
    public void send() throws InterruptedException {
        template.convertAndSend("sanyouTopic", "三友的java日记");
        Thread.sleep(60000);
    }

}

直接注入一个RocketMQTemplate，然后通过RocketMQTemplate发送消息。

运行结果如下：

处理消息:三友的java日记

的确消费到消息了。

原理

其实原理是一样的，只不过在SpringBoot中给封装了一层，让使用起来更加简单。

1、RocketMQTemplate构造代码

图片

所以从这可以看出，最终在构造RocketMQTemplate的时候，传入了一个DefaultMQProducer，所以可想而知，最终RocketMQTemplate发送消息也是通过DefaultMQProducer发送的。

2、@RocketMQMessageListener 注解处理

图片

从这可以看出，会为每一个加了@RocketMQMessageListener注解的对象创建一个DefaultMQPushConsumer，所以最终也是通过DefaultMQPushConsumer消费消息的。

至于监听器，是在这

图片

遍历每条消息，然后调用handleMessage，最终会调用实现了RocketMQListener的对象处理消息。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1242115.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

AI绘画工具汇总：免费、简单易上手

AI绘画工具汇总：免费、简单易上手

欢迎来到魔法宝库&#xff0c;传递AIGC的前沿知识&#xff0c;做有格调的分享❗ 喜欢的话记得点个关注吧&#xff01; 提到AI绘画&#xff0c;许多人通常会想到Midjourney和Stable Diffusion等工具&#xff0c;然而&#xff0c;这些工具对于新手而言门槛较高&#xff0c;不太友…
阅读更多...
20230511 Windows Ubuntu vscode remote-ssh 连接配置

20230511 Windows Ubuntu vscode remote-ssh 连接配置

参考 &#xff1a; VSCode SSH 连接远程ubuntu Linux 主机 VSCode通过Remote SSH扩展连接到内网Ubuntu主机 Ubuntu 安装 sudo apt-get install openssh-server vscode: 安装remote-ssh 插件 连接到服务器IP 免密登录的公钥密钥传递用filezillaUbuntu 和 Windows 文件互传 …
阅读更多...
INFLOW：用于检测隐藏服务器的反向网络流水印

INFLOW：用于检测隐藏服务器的反向网络流水印

文章信息 论文题目&#xff1a;INFLOW: Inverse Network Flow Watermarking for Detecting Hidden Servers 期刊&#xff08;会议&#xff09;&#xff1a;IEEE INFOCOM 2018 - IEEE Conference on Computer Communications 级别&#xff1a;CCF A 文章链接&#xff1a;https:…
阅读更多...
分布式篇---第二篇

分布式篇---第二篇

系列文章目录 文章目录 系列文章目录前言一、你知道哪些分布式事务解决方案？二、什么是二阶段提交？三、什么是三阶段提交？前言 前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家。点击跳转到网站，这篇文章男女通用，看懂了就去分享给你…
阅读更多...
AMEYA360：瑞萨面向高端工业传感器系统推出高精度模拟前端的32位RX MCU

AMEYA360：瑞萨面向高端工业传感器系统推出高精度模拟前端的32位RX MCU

全球半导体解决方案供应商瑞萨电子&#xff08;TSE&#xff1a;6723&#xff09;宣布面向高端工业传感器系统推出一款全新RX产品——RX23E-B&#xff0c;扩展32位微控制器&#xff08;MCU&#xff09;产品线。新产品作为广受欢迎的RX产品家族的一员&#xff0c;具有高精度模拟前…
阅读更多...
chatglm3部署使用

chatglm3部署使用

chatglm3部署使用 1.部署2.使用3.接入微信4.vue前端 1.部署 1.首先去github下载chatglm3代码。Huggingface下载模型一直失败&#xff0c;所以用阿里的魔塔社区下载。 git clone https://github.com/THUDM/ChatGLM3.git git clone https://www.modelscope.cn/ZhipuAI/chatglm3…
阅读更多...
2022-1-25 机器人运动规划方法综述 航空学报

2022-1-25 机器人运动规划方法综述 航空学报

论文PDF abstract 随着应用场景的日益复杂&#xff0c;机器人对旨在生成无碰撞路径&#xff08;轨迹&#xff09;的自主运动规划技术的需求也变得更加迫 切。虽然目前已产生了大量适应于不同场景的规划算法&#xff0c;但如何妥善地对现有成果进行归类&#xff0c;并分析不同…
阅读更多...
Linux MYSQL-5.7.23-rpm安装(附带安装包)

Linux MYSQL-5.7.23-rpm安装(附带安装包)

系统环境 OS 版本(Linux) CentOS-7-x86_64-Minimal-1511Java版本 jdk1.7及以上 卸载系统自带mariadb-lib rpm -qa|grep mariadbrpm -e mariadb-libs-5.5.68-1.el7.x86_64 --nodeps解压tar包 tar -xvf mysql*.tar以下加粗rpm包需要安装&#xff1a; mysql-community-common…
阅读更多...
求解Beamforming-SOCP（CVX求解）

求解Beamforming-SOCP（CVX求解）

时间&#xff1a;2023年11月23日14:00:16&#xff1a; 直接上代码&#xff08;辛苦两天才改出来的&#xff09; clear all; K 4; %user number N4; %base station number var1e-9; H []; %initialize H matrix for i1:Kh 1/sqrt(2*K)*mvnrnd(zeros(N,1),eye(N),1)1i/sqrt(2*…
阅读更多...
【运动规划】191 自适应跟踪kinodynamicrrt的路径

【运动规划】191 自适应跟踪kinodynamicrrt的路径

分层法&#xff1a; two layer approach 自适应控制&#xff0c;跟随轨迹。运动规划&#xff1a;扩展自由空间&#xff08;基于速度约束缩小自由空间&#xff09;为控制部分留余量&#xff0c;确保安全。 控制设计&#xff1a; 考虑平移和旋转&#xff0c;速度环控制&#xff…
阅读更多...
postgreSQL如何快速查询大表数据量

postgreSQL如何快速查询大表数据量

文章目录 场景方案结果 场景 我有一个非常大的表&#xff0c;估计几百万或者几千万。 我开始使用了 select count(*) from my_table_javapub 方式&#xff0c;查询非常慢。 如何解决&#xff1f;&#xff1f;&#xff1f; 方案 如果你需要更快地获取表中的行数&#xff0c…
阅读更多...
Python爬取京东商品销售数据进行数据分析示例代码，以口红为例

Python爬取京东商品销售数据进行数据分析示例代码，以口红为例

文章目录 一、准备工作驱动安装模块使用与介绍 二、流程解析三、完整代码四、效果展示关于Python技术储备一、Python所有方向的学习路线二、Python基础学习视频三、精品Python学习书籍四、Python工具包项目源码合集①Python工具包②Python实战案例③Python小游戏源码五、面试资…
阅读更多...
windows server-修改管理员登录密码-远程访问

windows server-修改管理员登录密码-远程访问

文章目录 1.打开服务器远程桌面2.修改计算机密码3.远程桌面登录 1.打开服务器远程桌面 允许远程连接到此计算机。 2.修改计算机密码 打开计算机管理>本地用户和组>修改管理员密码>重启。如下图所示&#xff1a; 3.远程桌面登录 打开远程桌面连接。输入ip&…
阅读更多...
关于APP备案的通知以及APP备案的常见问题

关于APP备案的通知以及APP备案的常见问题

前言 众所周知今年8月份&#xff0c;工信部出台了《工业和信息化部关于开展移动互联网应用程序备案工作的通知》&#xff0c;APP开发者的影晌是显而易见的。开发者需要按照要求提交相关材料进行备案&#xff0c;这无疑增加了开发者的时间和精力成本。虽然备案制度会增加开发者…
阅读更多...
计网（复习自用）

计网（复习自用）

计算机网络 1.概述 1.1概念 含义 计算机网络&#xff1a;是一个将分散的。具有独立功能的计算机系统&#xff0c;通过通信设备和线路连接起来&#xff0c;由功能完善的软件实现资源共享和信息传递的系统。 简单点说&#xff0c;计算机网络是互联的&#xff0c;自治的计算机集…
阅读更多...
为何设计师都在用这个原型样机资源网站？

为何设计师都在用这个原型样机资源网站？

谈论原型样机素材模板&#xff0c;这个话题对设计师来说如同老朋友一般熟悉。设计师们在创作完毕后&#xff0c;为了更淋漓尽致地展示他们的设计成果&#xff0c;通常会将其放置在真实的样机素材模板中。这种原型样机素材可以让设计作品迅速且清晰地呈现在真实环境中。找到一个…
阅读更多...
springboot_vue知识点

springboot_vue知识点

代码放到了仓库。 springboot_vue知识点 1.搭建1.vue2.springboot 2.前后端请求和响应的封装1.请求封装2.响应封装 3.增删改查1.查询2.分页3.新增和编辑4.删除 4.跨域和自定义异常5.JWT鉴权1.配置pom2.拦截前端请求的拦截器3.生成token并验证token4.登录后生成token5.前端获取…
阅读更多...
Spark---基于Standalone模式提交任务

Spark---基于Standalone模式提交任务

Standalone模式两种提交任务方式 一、Standalone-client提交任务方式 1、提交命令 ./spark-submit --master spark://mynode1:7077 --class org.apache.spark.examples.SparkPi ../examples/jars/spark-examples_2.11-2.3.1.jar 100 或者 ./spark-submit --master spark…
阅读更多...
centos7 系统keepalived 定时执行脚本

centos7 系统keepalived 定时执行脚本

安装keepalived yum install -y keepalived 修改配置文件 配置文件路径 /etc/keepalived 配置文件内容 global_defs {router_id localhost.localdomain # 访问到主机&#xff0c;本机的hostname&#xff0c;需要修改 }vrrp_script chk_http_port {script "/etc/kee…
阅读更多...
2022-4-11 南科大现代控制与最优估计

2022-4-11 南科大现代控制与最优估计

CLEAR_LAB B站视频 矩阵的分块矩阵操作 diagonal 对角阵 identity matrix 单位矩阵 矩阵克罗内克积
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023