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…
阅读更多...
AUTOSAR从入门到精通-【应用篇】基于 AUTOSAR 的辅助驾驶系统域控制器软件开发（续）

AUTOSAR从入门到精通-【应用篇】基于 AUTOSAR 的辅助驾驶系统域控制器软件开发（续）

目录 3.4 软件平台 CameraService 模块开发 3.4.1 中继模块 3.4.2 数据管理与调度模块
阅读更多...
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…
阅读更多...
Java基础十八（正则表达式 + 日期时间）

Java基础十八（正则表达式 + 日期时间）

1. 正则表达式 1.1 普通字符 字符描述示例[abc]匹配 […] 中所有字符[hlo] 匹配字符串 "hello world" 中所有的 h l o 字母[^ABC]匹配除了 […] 中所有字符[hlo] 匹配字符串 "hello world" 中除了 h l o 的所有字母[^a-z]匹配除了 […] 中所有字符[hlo] 匹…
阅读更多...
Sinkhorn算法

Sinkhorn算法

Sinkhorn算法 介绍示例代码 介绍 Sinkhorn算法是一种用于解决最优传输问题的迭代算法。最优传输问题是指在给定两个概率分布 μ \mu μ和 ν \nu ν的情况下&#xff0c;找到一个最优的转移方案&#xff0c;使得从 μ \mu μ到 ν \nu ν的转移成本最小。Sinkhorn算法通过迭代…
阅读更多...
C语言static关键字

C语言static关键字

目录 概述1. 局部变量1.1 普通局部变量1.2 静态局部变量 2.全局变量2.1 普通全局变量2.2 静态全局变量 3.函数4.静态块总结 概述 本文简单介绍一下static关键字在c语言中的应用。 1. 局部变量 1.1 普通局部变量 普通局部变量是在函数内部或代码块内部定义的变量。这些变量只…
阅读更多...
Docker file解析

Docker file解析

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

单片机IO模拟串口协议

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

在渗透测试中，探测到6379端口，你打算怎么利用

在渗透测试中&#xff0c;探测到6379端口&#xff0c;你打算怎么利用&#xff1f; 6379端口是Redis数据库的默认端口号。Redis是一种基于键值对的内存数据库&#xff0c;具有高性能和灵活的数据存储和检索功能。6379端口是Redis数据库的默认通信端口&#xff0c;用于连接到Red…
阅读更多...
Redis数据结构：Set类型全面解析

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

Set 类型是一个无序并唯一的键值集合&#xff0c;它的存储顺序不会按照插入的先后顺序进行存储。Redis 中集合是通过哈希表实现的&#xff0c;所以添加&#xff0c;删除&#xff0c;查找的复杂度都是 O(1)。相对于列表&#xff0c;集合也有两个特点&#xff1a;无序、不可重复 …
阅读更多...
从零开始学习 Java：简单易懂的入门指南之包装类（十九）

从零开始学习 Java：简单易懂的入门指南之包装类（十九）

包装类 包装类5.1 概述5.2 Integer类5.3 装箱与拆箱5.4 自动装箱与自动拆箱5.5 基本类型与字符串之间的转换基本类型转换为StringString转换成基本类型 5.6 底层原理 算法小题练习一&#xff1a;练习二&#xff1a;练习三&#xff1a;练习四&#xff1a;练习五&#xff1a; 包装…
阅读更多...
Java代码优化案例2：使用HashMap代替List进行数据查找

Java代码优化案例2：使用HashMap代替List进行数据查找

在开发过程中&#xff0c;我们经常需要在一个集合中查找某个元素。一种常见的做法是使用List来存储数据&#xff0c;然后通过循环遍历List来查找目标元素。然而&#xff0c;当数据量较大时&#xff0c;这种做法效率较低。我们可以通过使用HashMap来优这个过程。 1. 原始代码实…
阅读更多...
vue3 单一状态管理

vue3 单一状态管理

目录 1&#xff0c;问题2&#xff0c;单一状态管理 1&#xff0c;问题 有一个全局错误弹窗&#xff0c;当项目中有接口报错时&#xff0c;会显示错误信息。 思路&#xff1a;在 axios 的响应拦截器中&#xff0c;满足条件后打开错误弹窗。 问题&#xff1a;如何在 js 文件&a…
阅读更多...
数据库——redis介绍

数据库——redis介绍

文章目录 redis是什么&#xff1f;分布式缓存常见的技术选型方案有哪些&#xff1f;说一下 Redis 和 Memcached 的区别和共同点&#xff1f; redis是什么&#xff1f; 简单来说 Redis 就是一个使用 C 语言开发的数据库&#xff0c;不过与传统数据库不同的是 Redis 的数据是存在…
阅读更多...
浅析三维模型OBJ格式轻量化压缩文件大小的技术方法

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

浅析三维模型OBJ格式轻量化压缩文件大小的技术方法 在减小三维模型OBJ格式轻量化文件大小方面&#xff0c;有许多技术和方法可以使用。下面我将介绍一些常用的方法来减小OBJ文件的大小。 1、优化顶点数量&#xff1a;减少OBJ文件中的顶点数量是减小文件大小的一种有效方法。可…
阅读更多...
同一个服务器发布两个前端（网站）

同一个服务器发布两个前端（网站）

一开始怎么设置都是505&#xff0c;后来把网站文件的位置换到原已经发布成功的网站位置&#xff0c;就成功了。考虑应该是权限问题 server {listen 80;server_name localhost;# https配置参考 start#listen 443 ssl;# 证书直接存放 /docker/nginx/cert/ 目录下即…
阅读更多...
最新文章

Copyright @ 2022~2023