Linux部署RocketMQ并使用SpringBoot创建生产、消费者

  • 😜作           者：是江迪呀
  • ✒️本文关键词：RocketMQ、消息队列
  • ☀️每日   一言：在你心灰意冷、心烦意乱时也不要停下你的脚步！

在这里插入图片描述

一、前言

RocketMQ（Apache RocketMQ）是一种开源的分布式消息中间件系统，最初由阿里巴巴开发并捐赠给 Apache 基金会。它提供了可靠的、低延迟的消息传递能力，适用于构建大规模分布式系统中的消息通信。RocketMQ 主要用于解决分布式系统中异步通信、解耦、流量削峰等问题。下面让我们一起看下，如何在Linux上部署RocketMQ~

二、介绍RocketMQ

2.1 RocketMQ产生背景

随着业务规模的扩大，阿里巴巴面临着越来越多的分布式系统构建需求。为了解决这个问题，阿里巴巴集团于2012年推出的开源分布式消息中间件 —— RocketMQ。

2.1 RocketMQ作用

（1）异步通信和解耦： RocketMQ可以在不同的服务之间实现异步通信，解耦了服务之间的紧耦合关系，提高了系统的可维护性和可扩展性。
（2）流量削峰： RocketMQ支持消息积压和消费速率不匹配时的流量削峰功能，防止系统因突发流量而崩溃。
（3）实时数据同步： 用于将数据实时同步到不同的存储介质，保持数据的一致性。
（4）事件驱动架构： RocketMQ支持事件驱动的架构，使得系统能够更加敏捷地响应业务事件。

2.2 RocketMQ的组件

RocketMQ 的主要组件包括：
（1）Producer：负责发送消息到 RocketMQ 服务器。
（2）Broker：消息中转服务器，负责存储消息并提供消息的读写服务。
（3）Consumer：从 Broker 订阅并消费消息。
（4）Topic：消息的分类，Producer 发送消息到特定的 Topic，Consumer 订阅相应的 Topic。
（5）Tag： 对消息的进一步分类，可以用于 Consumer 进一步过滤消息。
（6）Message Queue： 每个 Topic 下可以分成多个 Message Queue，实现消息的分区和负载均衡。

2.3 RocketMQ的优缺点

（1）优点

  • 高吞吐量： RocketMQ具有高吞吐量的特点，适用于大量消息的处理。
  • 可靠性： RocketMQ通过消息的持久化存储和复制机制，确保消息不会丢失。
  • 低延迟： RocketMQ在消息传递过程中能够保持较低的延迟，适用于实时性要求较高的场景。
  • 灵活的消息模式： 支持发布-订阅和点对点两种消息模式，根据业务需求进行选择。
  • 水平扩展： 可以通过增加Broker节点来实现水平扩展，提高消息处理能力。

（2）缺点

  • 维护成本： RocketMQ需要维护多个组件，包括Producer、Broker和Consumer等，涉及到一定的运维成本。
  • 学习曲线： 对于新手来说，学习和理解RocketMQ的一些概念和配置可能需要一定的时间。
  • 一致性保障： 虽然RocketMQ通过复制机制保障了消息的可靠性，但在极端情况下可能会存在消息的重复传递或乱序问题。

三、 RocketMQ如何部署

3.1 下载

RocketMQ下载地址

3.2 上传、解压

上传文件到Linux有两种方式：

（1）上传

  • 通过rz命令
rz

你可以使用rz命令，在使用这个命令之前你必须确保linux已经安装了lrzsz，安装命令如下:

sudo apt-get update
sudo apt-get install lrzsz
  • 使用xftp
    这个我就不赘述了。
    在这里插入图片描述

（2）解压

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

如果没有安装unzip，需要安装一下：

// 查看 unzip 包的安装情况
yum list unzip
//没有安装时，使用命令安装 unzip
yum list unzipyum install unzip.x86_64

在这里插入图片描述

3.2 启动RocketMQ

RocketMQ的启动主要涉及到Namesrv（命名服务）和Broker（消息存储和消费者服务）两部分。要想启动RocketMQ，首先进入解压后的bin目录：

cd rocketmq-all-4.5.2-bin-release/bin

（1）启动Namesrv并设置输出日志位置

nohup sh mqnamesrv > namesrv.log 2>&1 &

（2）启动Broker并设置输出日志位置

nohup sh mqbroker -n localhost:9876 > broker.log 2>&1 &

查看是否启动：

jps

输出下面的内容说明启动成功了：

2931 NamesrvStartup
25599 Jps
25583 BrokerStartup

在启动Broker会出现失败问题，一般来说就是内存不足 ，RocketMq默认的虚拟机内存较大，因而启动失败，需要编辑如下两个配置文件，修改jvm的内存大小：

//编辑runbroker.sh和runserver.sh修改默认的JVM大小
vim runbroker.sh
vim runserver.sh

在这里插入图片描述
修改为：

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -xx:metaspaceSize=128m -XX:MaxMetaspaceSize=320m"

如果还是启动不了，需要将NameServer关闭，重新启动一下，同样是先进入bin目录，关闭命令如下：

sh mqshutdown namesrv

四、测试与关闭

4.1 测试

（1） 发送消息（生产者）

//设置环境变量
export NAMESRV_ADDR=localhost:9876
//使用安装包的demo发送消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

在这里插入图片描述
上面的信息就是RocketMQ的producer发送的消息。特点：启动发送完毕消息后就会停止。

（2） 接收消息（消费者）

//设置环境变量
export NAMESRV_ADDR=localhost:9876
//接收消息
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

在这里插入图片描述

4.2 关闭RocketMQ

//关闭namesrv
sh bin/mqshutdown namesrv
//关闭Broker
sh bin/mqshutdown broker

五、SpringBoot连接RocketMQ

5.1 引入依赖

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-thymeleaf</artifactId><version>2.3.5.RELEASE</version></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.5.2</version> </dependency>
</dependencies>

5.2 配置文件application.properties

# Name Server地址
rocketmq.name-server=your-nameserver-ip:9876
# 生产者组名
rocketmq.producer.group=my-producer-group
# 消费者组名
rocketmq.consumer.group=my-consumer-group

5.3 生产者

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;@Component
public class RocketMQProducer {@Value("${rocketmq.name-server}")private String nameServer;@Value("${rocketmq.producer.group}")private String producerGroup;public void sendMessage(String topic, String message) throws Exception {DefaultMQProducer producer = new DefaultMQProducer(producerGroup);producer.setNamesrvAddr(nameServer);producer.start();// 创建消息对象，设置消息内容org.apache.rocketmq.common.message.Message msg = new org.apache.rocketmq.common.message.Message(topic, message.getBytes());// 发送消息producer.send(msg);producer.shutdown();}
}

5.4 消费者

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;@Component
public class RocketMQConsumer {@Value("${rocketmq.name-server}")private String nameServer;@Value("${rocketmq.consumer.group}")private String consumerGroup;public void startConsumer(String topic) throws MQClientException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);consumer.setNamesrvAddr(nameServer);// 订阅主题和标签，可以根据需要进行过滤consumer.subscribe(topic, "*");// 注册消息监听器consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {for (org.apache.rocketmq.common.message.MessageExt msg : msgs) {System.out.println("Received message: " + new String(msg.getBody()));}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});consumer.start();}

5.5 启动类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;@SpringBootApplication
public class RocketMQDemoApplication {public static void main(String[] args) throws Exception {ConfigurableApplicationContext context = SpringApplication.run(RocketMQDemoApplication.class, args);RocketMQProducer producer = context.getBean(RocketMQProducer.class);producer.sendMessage("my-topic", "Hello, RocketMQ!");RocketMQConsumer consumer = context.getBean(RocketMQConsumer.class);consumer.startConsumer("my-topic");}
}

六、RocketMQ集群

上面所述的是单体RocketMQ，也能使用。但是如果你想要实现高可用在实际的业务场景中。RocketMQ大部分都不会单体存在，需要搭建集群来实现高可用。

有人已经写好了，而且很详细：传送门

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

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

使用Python爬虫采集网络热点

使用Python爬虫采集网络热点

在当今信息爆炸的时代&#xff0c;了解网络热搜词和热点事件对于我们保持时事敏感性和把握舆论动向非常重要。在本文中&#xff0c;我将与你分享使用Python爬虫采集网络热搜词和热点事件的方法&#xff0c;帮助你及时获取热门话题和热点新闻。 1. 网络热搜词采集 网络热搜词是人…
阅读更多...
JS-this知识点、面试题

JS-this知识点、面试题

一、this指向什么 1.简介 2.规则一&#xff1a;默认绑定 3.规则二&#xff1a;隐式绑定 4.规则四&#xff1a;new绑定 5.规则三&#xff1a;显式绑定 call、apply、bind 6.内置函数的绑定 7.规则优先级 8.this规则之外--es6剪头函数 9.ES6剪头函数this 二、This面试题 面试题…
阅读更多...
学习Linux的注意事项（使用经验；目录作用；服务器注意事项）

学习Linux的注意事项（使用经验；目录作用；服务器注意事项）

本篇分享学习Linux过程中的一些经验 文章目录 1. Linux系统的使用经验2. Linux各目录的作用3. 服务器注意事项 1. Linux系统的使用经验 Linux严格区分大小写Linux中所有内容以文件形式保存&#xff0c;包括硬件&#xff0c;Linux是以管理文件的方式操作硬件 硬盘文件是/dev/s…
阅读更多...
【深度学习】实验03 特征处理

【深度学习】实验03 特征处理

文章目录 特征处理标准化归一化正则化 特征处理 标准化 # 导入标准化库 from sklearn.preprocessing import StandardScalerfrom matplotlib import gridspec import numpy as np import matplotlib.pyplot as plt import warnings warnings.filterwarnings("ignore&quo…
阅读更多...
java练习8.100m小球落地

java练习8.100m小球落地

题目: 如一个小球从100米高度自由落下&#xff0c;每次落地后就反跳回原高度的一半。 那么求它在第10次落地时&#xff0c;共经过多少米&#xff1f;第10次反弹多高&#xff1f; public static void main(String[] args) {/*假如一个小球从100米高度自由落下&#xff0c;每次落…
阅读更多...
【Unity开发必备】100多个 Unity 学习网址 资源 收藏整理大全【持续更新】

【Unity开发必备】100多个 Unity 学习网址 资源 收藏整理大全【持续更新】

Unity 相关网站整理大全 众所周知&#xff0c;工欲善其事必先利其器&#xff0c;有一个好的工具可以让我们事半功倍&#xff0c;有一个好用的网站更是如此&#xff01; 但是好用的网站真的太多了&#xff0c;收藏夹都满满的(但是几乎没打开用过&#x1f601;)。 所以本文是对…
阅读更多...
服务器的介绍

服务器的介绍

1.服务器概述 1.1 服务器的基本概念 服务器是计算机的一种&#xff0c;是网络中为客户端计算机提供各种服务的高性能计算机&#xff1b; 服务器在网络操作系统的控制下&#xff0c;将与其相连的硬盘、磁带、 打印机及昂贵的专用通讯设备提供给网络上的客户站点共享&#xf…
阅读更多...
ISIS路由协议

ISIS路由协议

骨干区域与非骨干区域 凡是由级别2组建起来的邻居形成骨干区域&#xff1b;级别1就在非骨干区域&#xff0c;骨干区域有且只有一个&#xff0c;并且需要连续&#xff0c;ISIS在IP环境下目前不支持虚链路。 路由器级别 L1路由器只能建立L1的邻居&#xff1b;L2路由器只能建立L…
阅读更多...
软件测试的方法有哪些？

软件测试的方法有哪些？

软件测试 根据利用的被测对象信息的不同&#xff0c;可以将软件测试方法分为&#xff1a;黑盒测试、灰盒测试、白盒测试。 1、白盒测试 1&#xff09;概念&#xff1a;是依据被测软件分析程序内部构造&#xff0c;并根据内部构造分析用例&#xff0c;来对内部控制流程进行测试…
阅读更多...
【给自己挖个坑】三维视频重建(NSR技术)-KIRI Engine

【给自己挖个坑】三维视频重建(NSR技术)-KIRI Engine

文章目录 以下是我和AI的对话通过手机拍摄物体的视频&#xff0c;再根据视频生成三维模型&#xff0c;这个可实现吗我想开发类似上面的手机应用程序&#xff0c;如何开发呢 看了以上回答&#xff0c;还是洗洗睡吧NSR技术的实现原理是什么呢有案例吗我是名Java工程师&#xff0c…
阅读更多...
AI时代,程序员需要焦虑吗?

AI时代,程序员需要焦虑吗?

原文来自 微信公众号"互联网技术人进阶之路". 目录 前言一、程序员会被 AI 取代么&#xff1f;二、服务端开发尚难被 AI 取代三、服务端开发何去何从&#xff1f;四、业界首部体系化、全景式解读服务端开发的著作第一部分&#xff1a;服务端开发的技术和方法第二部分…
阅读更多...
docker高级（mysql主从复制）

docker高级（mysql主从复制）

数据库密码需要设置成自己的&#xff01;&#xff01;&#xff01; 1、创建容器master13307 #docker pulldocker run -p 13307:3306 --name mysql-master \ --privilegedtrue \ -v /mysql/mysql-master/log:/var/log/mysql \ -v /mysql/mysql-master/data:/var/lib/mysql \ -…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023