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.mzph.cn/news/55975.shtml

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

相关文章

聊一聊微前端框架的选型和实现 | 业务平台

聊一聊微前端框架的选型和实现 | 业务平台

一、项目背景 目前&#xff0c;我们开发维护的项目主要有 6 个&#xff0c;但是分别对应 PC 和 H5 两个端&#xff1a; 如上图所示&#xff0c;我们 6个项目最开始是一个一个进行开发维护的&#xff0c;但是到后期&#xff0c;这几个项目之间有的部分会有业务逻辑不同&#xff…
阅读更多...
BUUCTF [SWPU2019]Web1

BUUCTF [SWPU2019]Web1

​ 这是一道sql二次注入题目&#xff0c;但是注入点并不在登录处 注册一个用户然后登录 广告申请处进行sql注入 你会发现过滤了很多关键字 空格#information等等 这里用到了一些绕过技巧 使用 /**/ 代替空格 union/**/select/**/1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,1…
阅读更多...
c++11 标准模板（STL）（std::basic_ostringstream）（一）

c++11 标准模板（STL）（std::basic_ostringstream）（一）

定义于头文件 <sstream> template< class CharT, class Traits std::char_traits<CharT> > class basic_ostringstream;(C11 前)template< class CharT, class Traits std::char_traits<CharT>, class Allocator std::allo…
阅读更多...
LLaMA中ROPE位置编码实现源码解析

LLaMA中ROPE位置编码实现源码解析

1、Attention中q&#xff0c;经下式&#xff0c;生成新的q。m为句长length&#xff0c;d为embedding_dim/head θ i 1 1000 0 2 i d \theta_i\frac{1}{10000^\frac{2i}{d}} θi​10000d2i​1​ 2、LLaMA中RoPE源码 import torchdef precompute_freqs_cis(dim: int, end: i…
阅读更多...
uniapp 开发微信小程序使用echart的dataZoom属性缩放功能不生效！bug记录！

uniapp 开发微信小程序使用echart的dataZoom属性缩放功能不生效！bug记录！

在本项目中使用的是这个echart库 在项目中添加了dataZoom配置项但是不生效&#xff0c;突然想到微信小程序代码大小的限制&#xff0c;之前的echarts.js是定制的&#xff0c;有可能没有加dataZoom组件。故重新定制echarts.js。之前用的echarts版本是5.0.0&#xff0c;这次也是…
阅读更多...
Web Components

Web Components

Web Components标准非常重要的一个特性是&#xff0c;它使开发者能够将HTML页面的功能封装为custom elements&#xff08;自定义标签&#xff09;&#xff0c;可以使用CustomElementRegistry来管理自定义标签 <script>//1、创建自定义标签class NewElement extends HTML…
阅读更多...
Docker file解析

Docker file解析

文章目录 简介构建的三步骤Docker执行Dockerfile的大致流程DockerFile常用保留字指令创建第一个Dockerfile镜像的缓存特性 Docker file 解析 简介 Dockerfile是用来构建Docker镜像的文本文件&#xff0c;是由一条条构建镜像所需的指令和参数构成的脚本&#xff0c;记录了镜像构…
阅读更多...
单片机IO模拟串口协议

单片机IO模拟串口协议

一、前言 嵌入式硬件平台调试中常用的debug方法是看串口打印定位问题&#xff0c;但有时候会遇到单片机没有串口外设或者串口引脚被占用的情况&#xff0c;这时候也可以在代码里操作空闲的IO输出不同个数的脉冲来达到调试的效果&#xff0c;但是要用逻辑分析仪抓线逐个看波形比…
阅读更多...
Redis数据结构：Set类型全面解析

Redis数据结构：Set类型全面解析

Set 类型是一个无序并唯一的键值集合&#xff0c;它的存储顺序不会按照插入的先后顺序进行存储。Redis 中集合是通过哈希表实现的&#xff0c;所以添加&#xff0c;删除&#xff0c;查找的复杂度都是 O(1)。相对于列表&#xff0c;集合也有两个特点&#xff1a;无序、不可重复 …
阅读更多...
浅析三维模型OBJ格式轻量化压缩文件大小的技术方法

浅析三维模型OBJ格式轻量化压缩文件大小的技术方法

浅析三维模型OBJ格式轻量化压缩文件大小的技术方法 在减小三维模型OBJ格式轻量化文件大小方面&#xff0c;有许多技术和方法可以使用。下面我将介绍一些常用的方法来减小OBJ文件的大小。 1、优化顶点数量&#xff1a;减少OBJ文件中的顶点数量是减小文件大小的一种有效方法。可…
阅读更多...
【Springboot】| 从深入自动配置原理到实现 自定义Springboot starter

【Springboot】| 从深入自动配置原理到实现 自定义Springboot starter

目录 一. &#x1f981; 前言二. &#x1f981; Spring-boot starter 原理实现分析2.1 自动配置原理 三. &#x1f981; 操作实践3.1 项目场景3.2 搭建项目3.3 添加相关依赖3.4 删除一些不需要的东西3.5 发邮件工具类逻辑编写3.6 创建相关配置类3.7 创建 Spring.factories 文件…
阅读更多...
Dart PowerTCP Emulation for .NET Crack

Dart PowerTCP Emulation for .NET Crack

Dart PowerTCP Emulation for .NET Crack .NET CF上的PowerTCP Emulation为手持设备提供了高级的Internet通信组件。这些功能允许同步操作&#xff0c;这样可以消耗更少的资源&#xff0c;提供更大的灵活性&#xff0c;并生成易于维护的软件。带有.NET的PowerTCP仿真包括VT52、…
阅读更多...
服务器Linux系统配置mysql数据库主从自动备份

服务器Linux系统配置mysql数据库主从自动备份

服务器Linux系统配置mysql数据库主从自动备份 当数据内容越来越多的时候&#xff0c;数据库也变得越来越大了。如果不小心误删了&#xff0c;或者被黑主机了&#xff0c;那就什么都没有了。所以数据库的数据怎么能让它不丢失做到万无一失变得尤为重要&#xff01; 我是艾西&a…
阅读更多...
《机器学习核心技术》分类算法 - 决策树

《机器学习核心技术》分类算法 - 决策树

「作者主页」&#xff1a;士别三日wyx 「作者简介」&#xff1a;CSDN top100、阿里云博客专家、华为云享专家、网络安全领域优质创作者 「推荐专栏」&#xff1a;小白零基础《Python入门到精通》 决策树 1、决策树API2、决策时实际应用2.1、获取数据集2.2、划分数据集2.3、决策…
阅读更多...
微服务（rpc）

微服务（rpc）

微服务&#xff08;rpc&#xff09; 微服务必备的模块生产者消费者管理平台流量控制集群情况下如何做到流量监控 负载均衡服务发现和治理序列化传输序列化和反序列化 微服务是一种架构风格&#xff0c;将一个应用程序拆分为一组小型、独立的服务&#xff0c;每个服务都可以独立…
阅读更多...
深入理解Reactor模型的原理与应用

深入理解Reactor模型的原理与应用

1、什么是Reactor模型 Reactor意思是“反应堆”&#xff0c;是一种事件驱动机制。 和普通函数调用的不同之处在于&#xff1a;应用程序不是主动的调用某个 API 完成处理&#xff0c;而是恰恰相反&#xff0c;Reactor逆置了事件处理流程&#xff0c;应用程序需要提供相应的接口并…
阅读更多...
阿里云机器学习PAI全新推出特征平台 (Feature Store)，助力AI建模场景特征数据高效利用

阿里云机器学习PAI全新推出特征平台 (Feature Store)，助力AI建模场景特征数据高效利用

推荐算法与系统在全球范围内已得到广泛应用&#xff0c;为用户提供了更个性化和智能化的产品推荐体验。在推荐系统领域&#xff0c;AI建模中特征数据的复用、一致性等问题严重影响了建模效率。阿里云机器学习平台 PAI 推出特征平台&#xff08;PAI-FeatureStore&#xff09; 。…
阅读更多...
Python 密码破解指南：15~19

Python 密码破解指南：15~19

协议&#xff1a;CC BY-NC-SA 4.0 译者&#xff1a;飞龙 本文来自【OpenDocCN 饱和式翻译计划】&#xff0c;采用译后编辑&#xff08;MTPE&#xff09;流程来尽可能提升效率。 收割 SB 的人会被 SB 们封神&#xff0c;试图唤醒 SB 的人是 SB 眼中的 SB。——SB 第三定律 十五、…
阅读更多...
前端基础（Element、vxe-table组件库的使用）

前端基础（Element、vxe-table组件库的使用）

前言&#xff1a;在前端项目中&#xff0c;实际上&#xff0c;会用到组件库里的很多组件&#xff0c;本博客主要介绍Element、vxe-table这两个组件如何使用。 目录 Element 引入element 使用组件的步骤 使用对话框的示例代码 效果展示 vxe-table 引入vxe-table 成果展…
阅读更多...
深入理解Android消息机制的原理

深入理解Android消息机制的原理

Handler & Looper & MessageQueue关系简述 一个线程至多有一个looper&#xff1b;一个looper有一个mq&#xff1b;一个mq对应多个message&#xff1b;一个message对应多个handler。消息类型&#xff1a;同步、异步、同步屏障消息。无限循环&#xff1a;在队列中没有消…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023