SpringBoot+RocketMQ集群(dledger)部署完整学习笔记

文章目录

  • 前言
  • 一、单台集群部署
  • 二、多台集群部署
    • 1.修改配置
    • 2.dashboard修改
  • 三、整合springboot
    • 1.引入pom和修改yml
    • 2.编写消费者
    • 3.编写生产者
    • 4.测试效果
  • 总结

前言

RocketMQ集群方式有好几种
官网地址 https://rocketmq.apache.org/zh/docs/4.x/deployment/01deploy

  • 2m-2s-async：2主2从异步刷盘(吞吐量较大，但是消息可能丢失
  • 2m-2s-sync：2主2从同步刷盘(吞吐量会下降，但是消息更安全)
  • 2m-noslave ：2主无从(单点故障)，然后还可以直接配置broker.conf，进行单点环境配置
  • dledger：用来实现主从切换的。集群中的节点会基于Raft协议随机选举出一个leader，
    其他的就都是follower。通常正式环境都会采用这种方式来搭建集群。

dledger搭建参考文档 https://rocketmq.apache.org/zh/docs/4.x/bestPractice/02dledger

MQ安装部署请看这篇：https://blog.csdn.net/HBliucheng/article/details/135357998

搭建过程中踩过的坑也也会记录下来

一、单台集群部署

## 启动
nohup sh bin/dledger/fast-try.sh start
## 关闭
nohup sh bin/dledger/fast-try.sh stop

先启动 fast-try.sh start
启动时发现权限不足
nohup: 无法运行命令"bin/mqbroker": 权限不够
查看启动脚本

cat bin/dledger/fast-try.sh

在这里插入图片描述
那我们就修改下nohup 后面加上sh
修改后如下

function startNameserver() {export JAVA_OPT_EXT=" -Xms512m -Xmx512m  "nohup sh bin/mqnamesrv &
}function startBroker() {export JAVA_OPT_EXT=" -Xms1g -Xmx1g  "conf_name=$1nohup sh bin/mqbroker -c $conf_name &
}

再次启动发现可以了
在这里插入图片描述
执行命令 查看集群情况 BID =0的是主节点

sh bin/mqadmin clusterList -n 127.0.0.1:9876

在这里插入图片描述
再看看dashboarb
启动之前请先开放6个端口 如果还有端口访问不了的请自行开放出来

firewall-cmd --zone=public --add-port=30909/tcp --permanent
firewall-cmd --zone=public --add-port=30911/tcp --permanent
firewall-cmd --zone=public --add-port=30919/tcp --permanent
firewall-cmd --zone=public --add-port=30921/tcp --permanent
firewall-cmd --zone=public --add-port=30929/tcp --permanent
firewall-cmd --zone=public --add-port=30931/tcp --permanent### 如果不想一次次开放下面命令也可以
firewall-cmd --zone=public --add-port=30900-30930/tcp --permanent
## 重启防火墙
systemctl reload firewalld
## 查看开放的端口
firewall-cmd --list-ports
## 其它命令
### 关闭端口
firewall-cmd --zone=public --remove-port=30909/tcp --permanent

在这里插入图片描述

启动生产者和消费者再看 master消费一个
在这里插入图片描述
停止master

 lsof  -i:30911## 找到pid杀死 我的是118276kill  118276

在这里插入图片描述
在这里插入图片描述
我们再启动 被杀死的broker

nohup sh  bin/mqbroker -c conf/dledger/broker-n0.conf &

在这里插入图片描述
在这里插入图片描述
发现30911作为slave回来了

二、多台集群部署

先准备三台机器
192.168.141.101
192.168.141.102
192.168.141.103

1.修改配置

192.168.141.101修改如下
profile不修改也可以

vim /etc/profile
## 加入192.168.141.102 192.168.141.103 同理102 103也改成这样
export NAMESRV_ADDR=192.168.141.101:9876
source /etc/profile

修改 broker.conf 后面我们启动哪个就修改哪个 我是把 broker-n0.conf复制一份到broker.conf，也可以直接修改broker-n0.conf，启动时启动自己配置的conf文件就可以

cd /bsoft/mdt/rocketmq/rocketmq-4.8.0/rocketmq-4.8.0/conf/dledger
cp broker-n0.conf broker.conf
vim broker.conf 
## 修改的配置如下改动地方namesrvAddr dLegerPeers dLegerSelfId
brokerClusterName = RaftCluster
brokerName=RaftNode00
listenPort=30911
namesrvAddr=192.168.141.101:9876;192.168.141.102:9876;192.168.141.103:9876
storePathRootDir=/tmp/rmqstore/node00
storePathCommitLog=/tmp/rmqstore/node00/commitlog
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
dLegerPeers=n0-192.168.141.101:40911;n1-192.168.141.102:40911;n2-192.168.141.103:40911
## must be unique
dLegerSelfId=n0
sendMessageThreadPoolNums=16

192.168.141.102修改如下
profile不修改也可以

vim /etc/profile
## 加入192.168.141.102 192.168.141.103 同理102 103也改成这样
export NAMESRV_ADDR=92.168.141.102:9876
source /etc/profile


cd /bsoft/mdt/rocketmq/rocketmq-4.8.0/rocketmq-4.8.0/conf/dledger
cp broker-n0.conf broker.conf
vim broker.conf 
## 修改的配置如下改动地方namesrvAddr dLegerPeers dLegerSelfId
brokerClusterName = RaftCluster
brokerName=RaftNode00
listenPort=30911
namesrvAddr=192.168.141.101:9876;192.168.141.102:9876;192.168.141.103:9876
storePathRootDir=/tmp/rmqstore/node00
storePathCommitLog=/tmp/rmqstore/node00/commitlog
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
dLegerPeers=n0-192.168.141.101:40911;n1-192.168.141.102:40911;n2-192.168.141.103:40911
## must be unique
dLegerSelfId=n1
sendMessageThreadPoolNums=16

192.168.141.103修改如下
profile不修改也可以

vim /etc/profile
## 加入192.168.141.102 192.168.141.103 同理102 103也改成这样
export NAMESRV_ADDR=192.168.141.103:9876

cd /bsoft/mdt/rocketmq/rocketmq-4.8.0/rocketmq-4.8.0/conf/dledger
cp broker-n0.conf broker.conf
vim broker.conf 
## 修改的配置如下改动地方namesrvAddr dLegerPeers dLegerSelfId
brokerClusterName = RaftCluster
brokerName=RaftNode00
listenPort=30911
namesrvAddr=192.168.141.101:9876;192.168.141.102:9876;192.168.141.103:9876
storePathRootDir=/tmp/rmqstore/node00
storePathCommitLog=/tmp/rmqstore/node00/commitlog
enableDLegerCommitLog=true
dLegerGroup=RaftNode00
dLegerPeers=n0-192.168.141.101:40911;n1-192.168.141.102:40911;n2-192.168.141.103:40911
## must be unique
dLegerSelfId=n2
sendMessageThreadPoolNums=16

开放端口
每台机器都要开放

firewall-cmd --zone=public --add-port=30911/tcp --permanent
firewall-cmd --zone=public --add-port=40911/tcp --permanent
systemctl reload firewalld

如果还有端口没开放，请自行开放

启动
每台机器都要启动

cd /bsoft/mdt/rocketmq/rocketmq-4.8.0/rocketmq-4.8.0
nohup sh bin/mqnamesrv &
nohup sh  bin/mqbroker -c conf/dledger/broker.conf &

查看日志
在这里插入图片描述
发现未创建文件夹，创建文件夹

 mkdir -p  /tmp/rmqstore/node00/commitlog

## 关掉再启动
sh bin/mqshutdown broker
## 启动broker
nohup sh  bin/mqbroker -c conf/dledger/broker.conf &

查看集群情况

sh bin/mqadmin clusterList -n 127.0.0.1:9876

在这里插入图片描述

踩坑 这个值不要随便写，这里从0开始递增 ，不然选举会有问题
在这里插入图片描述

2.dashboard修改

修改配置

## 根据自己的服务器地址修改，注意中间是分号不是逗号
rocketmq.config.namesrvAddr=192.168.141.101:9876;192.168.141.102:9876;192.168.141.103:9876

启动访问
在这里插入图片描述
关闭master再查看集群情况，然后再重启，和前面的单机集群一样的，大家可自行测试

三、整合springboot

1.引入pom和修改yml

        <dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.2</version></dependency>
rocketmq:
# 集群中间以分号隔开name-server: 192.168.141.101:9876;192.168.141.102:9876;192.168.141.103:9876producer:group: my_group_test

2.编写消费者

package com.study.config.rocketmq;import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;import java.nio.charset.Charset;/*** @author: * @time: 2024/1/5 10:00*/
@Component
@RocketMQMessageListener(consumerGroup = "my_group_test",topic = "topic_test",selectorType = SelectorType.TAG,selectorExpression = "tagA")
@Slf4j
public class MQMsgListener implements RocketMQListener<MessageExt> {@Overridepublic void onMessage(MessageExt message) {String msgId = message.getMsgId();String msg = new String(message.getBody(), CharsetUtil.UTF_8);log.info("msgId={} msg={}",msgId,msg);}
}

@RocketMQMessageListener 注解参数如下：

  • topic: 消费者订阅的主题，即消费者将从这个主题中接收消息。
  • consumerGroup: 消费者组，多个消费者可以组成一个消费者组，共同从一个主题中接收消息。
  • consumeMode: 消费模式，指定消费者是以并发的方式接收消息还是以有序的方式接收消息。并发模式下，多个消费者可以同时接收消息；有序模式下，每个消费者按照消息的顺序依次接收消息。
  • messageModel: 消息模式，指定消息是以集群模式还是广播模式发送。集群模式下，消息将被发送到同一个主题的其中一个消费者；广播模式下，消息将被发送到主题的所有消费者。
  • selectorType: 过滤消息的方式，可以使用标签(Tag)或SQL92表达式(SQL92)来过滤消息。
  • selectorExpression: 过滤消息的表达式，可以使用标签(Tag)或SQL92表达式(SQL92)来指定过滤条件。
  • maxReconsumeTimes: 消息消费失败后，可被重复投递的最大次数。超过最大重试次数后，消息将被放入死信队列。
  • delayLevelWhenNextConsume: 并发模式的消息重试策略，指定消息消费失败后的重试延迟级别。设置为-1时，表示无需重试，直接将消息放入死信队列。

3.编写生产者

package com.study.controller;import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;/*** @author: * @time: 2024/1/5 10:17*/
@RestController
@RequestMapping("/mq")
@Slf4j
public class RocketMQProducerController {@ResourceRocketMQTemplate rocketMQTemplate;@PostMapping("/sendMessage")@ResponseBodypublic void sendMessage(String msg){rocketMQTemplate.asyncSend("topic_test", "hello mq", new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {log.info("msgId={}",sendResult.getMsgId());}@Overridepublic void onException(Throwable e) {e.printStackTrace();}});}
}

同步会有一点小问题，第一次启动不会消费，直接写成异步

4.测试效果

发现没有主题
在这里插入图片描述
追踪源码发现主题和过滤消息的表达式按照冒号分割
topic取第一位，过滤表达式取第二位
在这里插入图片描述

修改再试下

  rocketMQTemplate.asyncSend("topic_test:tagA", "hello mq", new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {log.info("msgId={}",sendResult.getMsgId());}@Overridepublic void onException(Throwable e) {e.printStackTrace();}});

发现可以了
在这里插入图片描述
前面写了个java客户端的消费者，改下消费组发现也可以消费

java客户端代码

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_test02";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();}
}

在这里插入图片描述
到此集群搭建完成，大家搭建过程中有遇到问题可以交流

总结

整个搭建过程不难就是有点繁琐，需要配置多台服务器

  • 其中配置brocker.conf时dLegerSelfId值这块要注意 ,dLegerSelfId是节点 id, 必须属于 dLegerPeers 中的一个；同 Group 内各个节点要唯一。这个值从0开始递增
    在这里插入图片描述

  • 同一台服务器上启动时先启动 namesrv 再启动 broker

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

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

相关文章

基于yolov2深度学习网络的车辆行人检测算法matlab仿真

基于yolov2深度学习网络的车辆行人检测算法matlab仿真

目录 1.算法运行效果图预览 2.算法运行软件版本 3.部分核心程序 4.算法理论概述 5.算法完整程序工程 1.算法运行效果图预览 2.算法运行软件版本 MATLAB2022a 3.部分核心程序 .......................................................... load yolov2.mat% 加载训练好的…
阅读更多...
【Spring Cloud】关于Nacos配置管理的详解介绍

【Spring Cloud】关于Nacos配置管理的详解介绍

&#x1f389;&#x1f389;欢迎来到我的CSDN主页&#xff01;&#x1f389;&#x1f389; &#x1f3c5;我是Java方文山&#xff0c;一个在CSDN分享笔记的博主。&#x1f4da;&#x1f4da; &#x1f31f;推荐给大家我的专栏《Spring Cloud》。&#x1f3af;&#x1f3af; &am…
阅读更多...
一文初步了解slam技术

一文初步了解slam技术

本文初步介绍slam技术&#xff0c;主要是slam技术的概述&#xff0c;涉及技术原理、应用场景、分类、以及各自优缺点&#xff0c;和slam技术的未来展望。 &#x1f3ac;个人简介&#xff1a;一个全栈工程师的升级之路&#xff01; &#x1f4cb;个人专栏&#xff1a;slam精进之…
阅读更多...
NetWorkX之社会网络分析

NetWorkX之社会网络分析

NetWorkX之社会网络分析 文章目录 NetWorkX之社会网络分析netwokx社会网络分析简介简单的案例使用networkx分析恋情关系总结 netwokx社会网络分析简介 networkx 是 Python 中一个非常强大的模块&#xff0c;用于创建、操作和研究图结构的网络。在社会网络分析中&#xff0c;它…
阅读更多...
k8s的集群调度

k8s的集群调度

scheduler&#xff1a;负责调度资源。把pod调度到node节点。通过算法调度。分为预算策略和优先策略 List-watch 在k8s集群当中通过List-watch的机制进行每个组件的协作。保持数据同步。可以实现每个组件之间的解耦。 通过kubectl来配置文件统一向集群内部的apiserver来发送命…
阅读更多...
HttpRunner辅助函数debugtalk.py

HttpRunner辅助函数debugtalk.py

辅助函数debugtalk.py Httprunner框架中&#xff0c;使用yaml或json文件进行用例描述&#xff0c;无法做一些复杂操作&#xff0c;如保存一些数据跨文件调用&#xff0c;或者实现一些复杂逻辑判断等&#xff0c;为了解决这个问题&#xff0c;引入了debugtalk.py辅助函数来进行一…
阅读更多...
基于Java+Springboot+Mybatis+Vue+微信小程序的轿车改装设计方案

基于Java+Springboot+Mybatis+Vue+微信小程序的轿车改装设计方案

微信小程序的轿车改装设计方案,用户可以自行在小程序中查看某型号轿车的零件&#xff0c;可以查看相关的汽车资源。 一、API1.1 SpringBoot框架搭建1.2 数据库设计1.3 实体映射创建Mapper1.4 接口封装1.5 常用字段类型 二、小程序2.1 项目创建2.2 首页2.3 产品中心页 三、管理端…
阅读更多...
7.27 SpringBoot项目实战 之 整合Swagger

7.27 SpringBoot项目实战 之 整合Swagger

文章目录 前言一、Maven依赖二、编写Swagger配置类三、编写接口配置3.1 控制器Controller 配置描述3.2 接口API 配置描述3.3 参数配置描述3.4 忽略API四、全局参数配置五、启用增强功能六、调试前言 在我们实现了那么多API以后，进入前后端联调阶段，需要给前端同学提供接口文…
阅读更多...
test mutation-03-变异测试 mujava Mutation 入门

test mutation-03-变异测试 mujava Mutation 入门

拓展阅读 开源 Auto generate mock data for java test.(便于 Java 测试自动生成对象信息) 开源 Junit performance rely on junit5 and jdk8.(java 性能测试框架。性能测试。压测。测试报告生成。) test 系统学习-04-test converate 测试覆盖率 jacoco 原理介绍 Java (muJ…
阅读更多...
【STM32】STM32学习笔记-串口发送和接收(27)

【STM32】STM32学习笔记-串口发送和接收(27)

00. 目录 文章目录 00. 目录01. 串口简介02. 串口相关API2.1 USART_Init2.2 USART_InitTypeDef2.3 USART_Cmd2.4 USART_SendData2.5 USART_ReceiveData 03. 串口发送接线图04. USB转串口模块05. 串口发送程序示例06. 串口发送支持printf07. 串口发送支持printf_v208.09.10. 01.…
阅读更多...
Vue3 使用 Teleport 封装 一个 Dialog

Vue3 使用 Teleport 封装 一个 Dialog

文章目录 什么是Teleport ？用法：1. 通过 to 指定传送的位置2. 禁用 teleport3. 共享一个 Teleport封装一个Dialog效果：什么是Teleport ？ 是一个内置组件，它可以将一个组件内部的一部分模板“传送”到该组件的 DOM 结构外层的位置去。 简单的说，Telep
阅读更多...
NVIDIA深入理解之pynvml库

NVIDIA深入理解之pynvml库

一、前言 写在前面 该文章是对我之前文章《Fedora上安装NVIDIA闭源显卡驱动》的一个拓展&#xff0c;正好寒假闲的没事干不如加深一下对NVIDIA的了解。Python是当前非常流行的一门编程语言&#xff0c;它以kiss为设计思想&#xff0c;能封装就能封装&#xff0c;给用户提供比…
阅读更多...
【Linux Shell】11. 输入/输出 重定向

【Linux Shell】11. 输入/输出 重定向

文章目录 【 1. 重定向简介 】【 2. 输出重定向 】【 3. 输入重定向 】【 4. Here Document 】【 5. /dev/null 文件 】 【 1. 重定向简介 】 大多数 UNIX 系统命令从终端接受输入并将所产生的输出发送回​​到原来输入的终端。一个命令通常从标准输入的地方读取输入&#xff…
阅读更多...
信号的互相关计算及时延估计

信号的互相关计算及时延估计

1. 信号的互相关计算 互相关反映向量x和移位&#xff08;滞后&#xff09;向量y之间的相似性。 最直观的解释是&#xff1a;互相关的作用是为了找到信号在哪一时刻与另一信号最像&#xff08;另一信号为本身时就是自相关&#xff09;&#xff01; 滑动求互相关&#xff08;图…
阅读更多...
海康威视摄像头+服务器+录像机配置校园围墙安全侦测区域入侵侦测+越界侦测

海康威视摄像头+服务器+录像机配置校园围墙安全侦测区域入侵侦测+越界侦测

一、适用场景 1、校园内&#xff0c;防止课外时间翻越围墙到校外、从校外翻越围墙到校内&#xff1b; 2、通过服务器摄像头的侦测功能及时抓图保存&#xff0c;为不安全因素提供数字化依据&#xff1b; 3、网络录像机保存监控视频&#xff0c;服务器保存抓拍到的入侵与越界&am…
阅读更多...
【React】02-如何理解React通过对DOM的模拟，最大限度地减少与DOM的交互

【React】02-如何理解React通过对DOM的模拟，最大限度地减少与DOM的交互

如何理解React通过对DOM的模拟&#xff0c;最大限度地减少与DOM的交互 背景分析关于虚拟DOM 背景 在学习React的过程中&#xff0c;发现很多文档上关于React的高效都有这么一句话的描述——React通过对DOM的模拟&#xff0c;最大限度地减少与DOM的交互&#xff0c;对于我这种前…
阅读更多...
mysql服务多实例运行

mysql服务多实例运行

1、官网下载mysql安装包 https://downloads.mysql.com/archives/community/ 2、解压安装包 tar -zxvf mysql-8.1.0-linux-glibc2.28-aarch64.tar.xz -C /usr/localmv /usr/local/mysql-8.1.0-linux-glibc2.28-aarch64 /usr/local/mysql 3、创建mysql用户组 groupadd…
阅读更多...
Java面试汇总——redis篇

Java面试汇总——redis篇

1、什么是缓存穿透 ? 怎么解决 ? 缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在&#xff0c;这样缓存就形同虚设&#xff08;只有数据库查到了&#xff0c;才会让redis缓存&#xff0c;但现在的问题是查不到&#xff09;&#xff0c;会频繁的去访问数据库。 解决…
阅读更多...
这款Web剪藏工具绝了，支持10+平台内容剪辑同步！

这款Web剪藏工具绝了，支持10+平台内容剪辑同步！

前言 Web Clipper 是一个开源项目&#xff0c;旨在帮助用户轻松地保存和组织网页内容。它可以作为浏览器插件安装到常见的浏览器中&#xff0c;如Chrome、Firefox 等&#xff0c;用户可以使用它来保存网页、截取文章、添加标签和注释等操作&#xff0c;从而方便地管理和分享自…
阅读更多...
任务调度中心

任务调度中心

可以服务器配置和权限&#xff0c;分配任务执行。当服务器下线后&#xff0c;任务会被在线服务器接管&#xff0c;当重新上线后会在次执行任务。接管任务的服务器会释放任务。调度过程的实现&#xff0c;可以二次开发。基于 netty tcp 通信开发。 下载地址&#xff1a; http:/…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023