RocketMQ单机部署完整学习笔记

文章目录

  • 前言
  • 一、RocketMQ是什么？
  • 二、使用步骤
    • 1.安装MQ
      • 1.安装JDK
      • 2.安装mq
      • 3.MQ配置(核心)
    • 2.搭建可视化dashboard
      • 1.下载源码
      • 2.修改配置
      • 3.启动
    • 3.整合java
      • 1.生产者
      • 2.消费者
      • 3.启动生产者
      • 4.启动消费者
      • 5.dashboard添加消费组
  • 三、总结
    • 全部的配置

前言

本文是基于4.X版本RocketMQ，从MQ的搭建，消息推送和消费以及dashboard的使用

一、RocketMQ是什么？

参考文档 https://rocketmq.apache.org/zh/docs/4.x/
重点角色如下

  • Producer：消息的发送者；举例：发信者
  • Consumer：消息接收者；举例：收信者
  • Broker：暂存和传输消息；举例：邮局
  • NameServer：管理Broker；举例：各个邮局的管理机构
  • Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息;举例
  • tag：消息标签，方便服务器过滤使用

二、使用步骤

1.安装MQ

首先安装jdk 再安装mq

1.安装JDK

  1. 查看Linux系统是否有自带的jdk
java -version

如果有 输入 rpm -qa | grep java 检测jdk的安装包
接着进行一个个删除包，输入：rpm -e --nodeps +包名
最后再次：rpm -qa | grep java检查是否删除完即可

  1. 下载jdk

https://www.oracle.com/java/technologies/downloads/#java8
资源连接地址(jdk和yarn)
https://download.csdn.net/download/HBliucheng/88696153
3. 上传jdk到linux服务器
在这里插入图片描述

  1. 解压jdk
tar -zvxf jdk-8u241-linux-x64..tar.gz
  1. 配置环境变量
    用vim /etc/profile进入编辑状态，加入下边这段配置
    JAVA_HOME 根据自己的解压路径来写
JAVA_HOME=/usr/local/jdk/jdk1.8.0_241
JRE_HOME=$JAVA_HOME/jre
PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
export JAVA_HOME JRE_HOME PATH CLASSPATH
  1. 重新加载配置
source /etc/profile
  1. 进行测试
java -version

在这里插入图片描述

2.安装mq

  1. 下载mq
    连接 https://archive.apache.org/dist/rocketmq/4.9.4/rocketmq-all-4.9.4-bin-release.zip

  2. 解压上传
    我下载的是公司的mq是4.8的，官网链接给的是根4.9的，这个问题不大，不影响

目录结构如下
在这里插入图片描述

在这里插入图片描述
在这里插入图片描述

  1. 配置环境变量
vim /etc/profile
# 在末尾加入下面配置 路径和自己解压的mq路径一直 上一步有截图
export ROCKETMQ_HOME=/bsoft/mdt/rocketmq/rocketmq-4.8.0/rocketmq-4.8.0
# 使环境变量生效
source /etc/profile

3.MQ配置(核心)

这一步很重要，配置完这里，那mq就算部署好了

  1. 修改runserver.sh
    默认配置比较大，修改启动大小
## cd /bin
vim runserver.sh
## 修改启动大小
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
  1. 启动服务nameserver
## 启动
nohup sh bin/mqnamesrv &
## 关闭
sh bin/mqshutdown namesrv

出现下面就算启动成功了
The Name Server boot success. serializeType=JSON
注意目录别进错了
在这里插入图片描述

  1. 指定NameServer地址
    相当于 broker注册到nameserver上
vim /etc/profile
# 在末尾加入下面配置 有多个时以分号隔开，这个是集群时使用的 mq端口默认是9876 
# 192.168.141.101是服务器地址
export NAMESRV_ADDR=192.168.141.101:9876
# 使环境变量生效
source /etc/profile
  1. 修改 runbroker.sh

修改启动参数

vim runbroker.sh
## 
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m"
  1. 修改broker.conf
    重要，核心配置，以后关于mq服务的配置都在这里
    conf目录下
    2m-2s-async
    2m-2s-sync
    2m-noslave
    dledger
    这四个目录是集群配置时会用到，我们这里是单机的先不管
vim broker.confbrokerClusterName = DefaultCluster
brokerName = broker-a
#brokerid,0就表示是Master，>0的都是表示
brokerId = 0
# 这个就是第三三步配置的export NAMESRV_ADDR=192.168.141.101:9876 多个以分号分割
namesrvAddr=192.168.141.101:9876
#如果是多网卡的机器，比如云服务器，那么需要在broker.conf中增加brokerIP1属性，
#指定所在机器的外网网卡地址
brokerIP1=192.168.141.101
#对外服务的监听端口
listenPort=10911
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
traceTopicEnable=true
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true

在这里插入图片描述
在这里插入图片描述

  1. 启动broker
    进入bin目录
    注意 -c 请加上，不加后面客户端使用时可能会出问题
#启动
nohup sh  bin/mqbroker -c conf/broker.conf &
# 关闭
sh bin/mqshutdown broker

查看nohup.out
出现这样的就说明启动成功了
The broker[broker-a, 192.168.141.101:10911] boot success. serializeType=JSON and name server is 192.168.141.101:9876

在这里插入图片描述

  1. 到此mq服务已启动完成
jps

在这里插入图片描述

  1. 查看日志
## 查看namesrv日志tailf /root/logs/rocketmqlogs/namesrv.log ## broker日志tailf /root/logs/rocketmqlogs/broker.log

其实前面启动的时候查看这里的日志也可以看是否启动没
停掉时先停broker 再停namesrv 启动先启 namesrv 再启动broker 因为broker 需要注册到namesrv 上

  1. 发送和接收消息

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

在这里插入图片描述

在这里插入图片描述

sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

在这里插入图片描述

2.搭建可视化dashboard

1.下载源码

点击下面直接下载4.X的
https://github.com/apache/rocketmq-dashboard/tree/release-1.0.0
说下
现在MQ已经到5.X了，但是现在还保留着4的,分支下拉到最后可以看到一个relaese-1.0.0
这个就是4.X用的，下载下来后解压
切记版本要对上，不然你和我一样折腾个一两天

在这里插入图片描述

2.修改配置

主要改下这个

rocketmq.config.namesrvAddr=192.168.141.101:9876

如果自己端口需要修改也可以，我是改成了8078
在这里插入图片描述

3.启动

在这里插入图片描述
访问 http://localhost:8078
在这里插入图片描述

3.整合java

1.生产者

直接上代码

package com.bsoft;import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;import java.io.IOException;/*** @author:* @time: 2023/12/29 15:40*/
public class MQPublisher {private final static String nameServer = "192.168.141.101:9876";private final static String producerGroup = "my_group";private final static String consumerGroup = "my_group";private final static String topic = "topic_test";public static void main(String[] args) throws IOException {// 初始化一个producer并设置Producer group nameDefaultMQProducer producer = new DefaultMQProducer(producerGroup);try {// 设置NameServer地址producer.setNamesrvAddr(nameServer);// 启动producerproducer.start();// 创建一条消息，并指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤Message msg = new Message(topic, "tagB", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));// 异步发送消息, 发送结果通过callback返回给客户端producer.send(msg, new SendCallback() {public void onSuccess(SendResult sendResult) {System.out.printf("OK %s %n",sendResult.getMsgId());}public void onException(Throwable e) {System.out.printf("Exception %s %n", e);e.printStackTrace();}},10000);} catch (Exception e) {e.printStackTrace();}System.in.read();}
}

2.消费者

package com.bsoft;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;/*** @author: liucheng* @time: 2023/12/29 15:39*/
public class MQConsumer {private final static String nameServer = "192.168.141.101:9876";private final static String consumerGroup = "my_group_test";private final static String topic = "topic_test";public static void main(String[] args) throws MQClientException, IOException, InterruptedException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup,true);// 设置NameServer的地址consumer.setNamesrvAddr(nameServer);// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息consumer.subscribe(topic, "tagE");// 注册回调实现类来处理从broker拉取回来的消息consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);msgs.forEach((msg)->{byte[] body = msg.getBody();String s = new String(body, Charset.defaultCharset());System.out.println("msg=================> " +s);});// 标记该消息已经被成功消费return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动消费者实例consumer.start();System.out.printf("Consumer Started......");
//        Thread.sleep(5000);
//        consumer.shutdown();System.in.read();}
}

3.启动生产者

在这里插入图片描述

在这里插入图片描述

查看详情
在这里插入图片描述

4.启动消费者

package com.bsoft;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;import java.io.IOException;
import java.nio.charset.Charset;
import java.util.List;/*** @author: liucheng* @time: 2023/12/29 15:39*/
public class MQConsumer {private final static String nameServer = "192.168.141.101:9876";private final static String consumerGroup = "my_group_test";private final static String topic = "topic_test";public static void main(String[] args) throws MQClientException, IOException, InterruptedException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);// 设置NameServer的地址consumer.setNamesrvAddr(nameServer);// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息consumer.subscribe(topic, "tagA");// 注册回调实现类来处理从broker拉取回来的消息consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);msgs.forEach((msg)->{byte[] body = msg.getBody();String s = new String(body, Charset.defaultCharset());System.out.println("msg=================> " +s);});// 标记该消息已经被成功消费return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});// 启动消费者实例consumer.start();System.out.printf("Consumer Started......");
//        Thread.sleep(5000);
//        consumer.shutdown();System.in.read();}
}

在这里插入图片描述
没日志打印，但是又显示消费了
在这里插入图片描述

5.dashboard添加消费组

查看消费组，发现没有，那添加一个吧
在这里插入图片描述
在这里插入图片描述
重启生产者发现可以消费

对于消费组还是得在dashboard创建好了再去写代码，有的说能够改配置能直接创建，试过了，没生效，先这样吧，有好的方法或搭建过程中遇到什么问题可以多多交流
补充下：官网说的是先创建再发送，还说一点，这个autoCreateTopicEnable 配置有可能会有问题，
在这里插入图片描述

三、总结

整个搭建过程踩了不少坑，比如
版本的不一致导致部分功能一直报错；
启动brocker时未指定实例文件没有加-c来启动导致部署失败；
消费组未在dashboard创建时代码中不显示消费信息
关于5.x无法打包的问题是因为缺少yarn-1.22.10.tar.gz 这个已经上传到jdk那个资源下了，把这个复制到
{path}\maven\repository\com\github\eirslett\yarn\1.22.10再打包即可

全部的配置

#  cat /etc/profileJAVA_HOME=/usr/local/jdk/jdk1.8.0_241
JRE_HOME=$JAVA_HOME/jre
PATH=$PATH:$JAVA_HOME/bin:$JRE_HOME/bin
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib
export JAVA_HOME JRE_HOME PATH CLASSPATH
export ROCKETMQ_HOME=/bsoft/mdt/rocketmq/rocketmq-4.8.0/rocketmq-4.8.0
export NAMESRV_ADDR=192.168.141.101:9876
#  cat runbroker.sh JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m"
# cat runserver.sh 
JAVA_OPT="${JAVA_OPT} -server -Xms512m -Xmx512m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"# cat broker.conf 
brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
namesrvAddr=192.168.141.101:9876
brokerIP1=192.168.141.101
listenPort=10911
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
traceTopicEnable=true
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true

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

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

相关文章

LeetCode 2221. 数组的三角和

LeetCode 2221. 数组的三角和

文章目录 1. 题目 2. 解题 1. 题目 给你一个下标从 0 开始的整数数组 nums &#xff0c;其中 nums[i] 是 0 到 9 之间&#xff08;两者都包含&#xff09;的一个数字。 nums 的 三角和 是执行以下操作以后最后剩下元素的值&#xff1a; num…
阅读更多...
I.MX6ULL开发笔记（二）——硬件外设操作

I.MX6ULL开发笔记（二）——硬件外设操作

0x01 点亮第一个RGB灯 在文章http://t.csdnimg.cn/EGWt9中有介绍Linux下文件目录&#xff0c;那么在Linux系统下&#xff0c;RGB灯也是一个设备&#xff0c;所以我们需要到/sys目录下去操作这个设备。 之后&#xff0c;我们进入到class目录&#xff0c;这里挂载着开发板上的外…
阅读更多...
Mongodb删除操作中字符序对结果的影响

Mongodb删除操作中字符序对结果的影响

本文还是要从删除操作的语法说起。 db.collection.deleteMany(<filter>,{writeConcern: <document>,collation: <document>,hint: <document|string>} ) 删除语法中&#xff0c;可以指定数据写入策略&#xff0c;字符序和使用的索引字段。 字符序&a…
阅读更多...
MySQL第四战：视图以及常见面试题（上）

MySQL第四战：视图以及常见面试题（上）

目录 目录&#xff1a; 一.视图 1.介绍什么是视图 2.视图的语法 语法讲解 实例操作 二.MySQL面试题 1.SQL脚本 2.面试题实战 三.思维导图 目录&#xff1a; 随着数字化时代的飞速发展&#xff0c;数据库技术&#xff0c;特别是MySQL&#xff0c;已经成为IT领域中不可…
阅读更多...
Mybatis简易搭建并查询数据库表内所有数据

Mybatis简易搭建并查询数据库表内所有数据

搭建步骤 1.在mysql中创建user表&#xff0c;添加数据2.创建maven模块&#xff0c;导入坐标3.在资源文件夹中导入需要用到的logback.xml4.编写MyBatis 核心配置文件->替换连接信息 解决硬编码问题5.编写 SQL映射文件->统一管理sgl语句&#xff0c;解决硬编码问题6.编码定…
阅读更多...
HackTheBox - Medium - Linux - Ambassador

HackTheBox - Medium - Linux - Ambassador

Ambassador Ambassador 是一台中等难度的 Linux 机器&#xff0c;用于解决硬编码的明文凭据留在旧版本代码中的问题。首先&#xff0c;“Grafana”CVE &#xff08;“CVE-2021-43798”&#xff09; 用于读取目标上的任意文件。在研究了服务的常见配置方式后&#xff0c;将在其…
阅读更多...
【管理篇 / 登录】❀ 06. macOS下使用USB配置线登录 ❀ FortiGate 防火墙

【管理篇 / 登录】❀ 06. macOS下使用USB配置线登录 ❀ FortiGate 防火墙

【简介】飞塔防火墙上都会配有CONSOLE接口&#xff0c;包装里都会配置一根USB配置线&#xff0c;通过这个接口和这根线&#xff0c;我们可以用命令的方式登录飞塔防火墙。随着苹果电脑的普及&#xff0c;我们来学习如何在macOS中使用USB配置线登录飞塔防火墙。 早期飞塔防火墙包…
阅读更多...
useContext

useContext

可以跨组件传值 其实主要的就是三步 1、const xxx React.createContext();创建一个context 2、<xxx.Provider value{{ num, setNum }}>父组件设置要传递的值 3、const { num, setNum } React.useContext(xxx);子组件下使用 特点&#xff1a; 1、可以有多个xxx.Pr…
阅读更多...
网络优化篇(一)---------TCP重传性能优化

网络优化篇(一)---------TCP重传性能优化

本文通过一个TCP重传优化的实际问题，详细讲解问题的分析、定位、优化过程。 通过本文你将学到: 如何通过linux命令和/proc文件系统分析TCP性能数据如何通过linux命令和netlink api分析某个具体的TCP连接的性能数据如何通过bcc工具分析TCP性能数据如何通过调整系统参数优化TCP重…
阅读更多...
第 121 场 LeetCode 双周赛题解

第 121 场 LeetCode 双周赛题解

A 大于等于顺序前缀和的最小缺失整数 模拟&#xff1a;先求最长顺序前缀的和 s s s &#xff0c;然后从 s s s 开始找没有出现在 n u m s nums nums 中的最小整数 class Solution { public:int missingInteger(vector<int> &nums) {unordered_set<int> vis(…
阅读更多...
[C#]使用DlibDotNet人脸检测人脸68特征点识别人脸5特征点识别人脸对齐人脸比对FaceMesh

[C#]使用DlibDotNet人脸检测人脸68特征点识别人脸5特征点识别人脸对齐人脸比对FaceMesh

【官方框架地址】 https://github.com/takuya-takeuchi/DlibDotNet 【算法介绍】 DlibDotNet是一个开源的.NET库&#xff0c;用于实现机器学习和计算机视觉应用。它基于C库dlib&#xff0c;通过C/CLI封装了dlib的所有功能&#xff0c;为.NET开发者提供了简单易用的API。以下是…
阅读更多...
蓝桥杯练习题（一）

蓝桥杯练习题（一）

&#x1f4d1;前言 本文主要是【算法】——蓝桥杯练习题&#xff08;一&#xff09;的文章&#xff0c;如果有什么需要改进的地方还请大佬指出⛺️ &#x1f3ac;作者简介&#xff1a;大家好&#xff0c;我是听风与他&#x1f947; ☁️博客首页&#xff1a;CSDN主页听风与他 …
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023