当前位置： 代码网 > it编程>数据库>大数据 > RabbitMQ（集群相关部署）

RabbitMQ（集群相关部署）

2024年08月01日 • 大数据 •我要评论
在centos8 服务器，3台服务器，分别进行MQ的集群步骤

rabbitmq

集群部署

环境准备：阿里云centos8 服务器，3台服务器，分别进行安装；

下载erlang

erlang和rabbitmq版本对照：https://www.rabbitmq.com/which-erlang.html

  1. 创建yum库配置文件
vim /etc/yum.repos.d/rabbitmq.repo
  1. 内容
# 内容 - 来自官方文档：https://www.rabbitmq.com/docs/install-rpm
# in /etc/yum.repos.d/rabbitmq.repo
##
## zero dependency erlang rpm
##
[modern-erlang]
name=modern-erlang-el8
# uses a cloudsmith mirror @ yum.novemberain.com in addition to its cloudsmith upstream.
# unlike cloudsmith, the mirror does not have any traffic quotas
baseurl=https://yum1.novemberain.com/erlang/el/8/$basearch
        https://yum2.novemberain.com/erlang/el/8/$basearch
        https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-erlang/rpm/el/8/$basearch
repo_gpgcheck=1
enabled=1
gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.e495bb49cc4bbe5b.key
gpgcheck=1
sslverify=1
sslcacert=/etc/pki/tls/certs/ca-bundle.crt
metadata_expire=300
pkg_gpgcheck=1
autorefresh=1
type=rpm-md

[modern-erlang-noarch]
name=modern-erlang-el8-noarch
# uses a cloudsmith mirror @ yum.novemberain.com.
# unlike cloudsmith, it does not have any traffic quotas
baseurl=https://yum1.novemberain.com/erlang/el/8/noarch
        https://yum2.novemberain.com/erlang/el/8/noarch
        https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-erlang/rpm/el/8/noarch
repo_gpgcheck=1
enabled=1
gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.e495bb49cc4bbe5b.key
       https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc
gpgcheck=1
sslverify=1
sslcacert=/etc/pki/tls/certs/ca-bundle.crt
metadata_expire=300
pkg_gpgcheck=1
autorefresh=1
type=rpm-md

[modern-erlang-source]
name=modern-erlang-el8-source
# uses a cloudsmith mirror @ yum.novemberain.com.
# unlike cloudsmith, it does not have any traffic quotas
baseurl=https://yum1.novemberain.com/erlang/el/8/srpms
        https://yum2.novemberain.com/erlang/el/8/srpms
        https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-erlang/rpm/el/8/srpms
repo_gpgcheck=1
enabled=1
gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.e495bb49cc4bbe5b.key
       https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc
gpgcheck=1
sslverify=1
sslcacert=/etc/pki/tls/certs/ca-bundle.crt
metadata_expire=300
pkg_gpgcheck=1
autorefresh=1


##
## rabbitmq server
##

[rabbitmq-el8]
name=rabbitmq-el8
baseurl=https://yum2.novemberain.com/rabbitmq/el/8/$basearch
        https://yum1.novemberain.com/rabbitmq/el/8/$basearch
        https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/rpm/el/8/$basearch
repo_gpgcheck=1
enabled=1
# cloudsmith's repository key and rabbitmq package signing key
gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9f4587f226208342.key
       https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc
gpgcheck=1
sslverify=1
sslcacert=/etc/pki/tls/certs/ca-bundle.crt
metadata_expire=300
pkg_gpgcheck=1
autorefresh=1
type=rpm-md

[rabbitmq-el8-noarch]
name=rabbitmq-el8-noarch
baseurl=https://yum2.novemberain.com/rabbitmq/el/8/noarch
        https://yum1.novemberain.com/rabbitmq/el/8/noarch
        https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/rpm/el/8/noarch
repo_gpgcheck=1
enabled=1
# cloudsmith's repository key and rabbitmq package signing key
gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9f4587f226208342.key
       https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc
gpgcheck=1
sslverify=1
sslcacert=/etc/pki/tls/certs/ca-bundle.crt
metadata_expire=300
pkg_gpgcheck=1
autorefresh=1
type=rpm-md

[rabbitmq-el8-source]
name=rabbitmq-el8-source
baseurl=https://yum2.novemberain.com/rabbitmq/el/8/srpms
        https://yum1.novemberain.com/rabbitmq/el/8/srpms
        https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/rpm/el/8/srpms
repo_gpgcheck=1
enabled=1
gpgkey=https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9f4587f226208342.key
gpgcheck=0
sslverify=1
sslcacert=/etc/pki/tls/certs/ca-bundle.crt
metadata_expire=300
pkg_gpgcheck=1
autorefresh=1
type=rpm-md
  1. 更新yum库
# --nobest表示所需安装包即使不是最佳选择也接受
yum update -y --nobest

# 还需关闭防火墙
systemctl stop firewalld
# 开机关闭防火墙：
systemctl disable firewalld
# 查看防火墙状态：
systemctl status firewalld
  1. 进行安装
yum install -y erlang

安装rabbitmq

erlang和rabbitmq版本对照：https://www.rabbitmq.com/which-erlang.html

# 导入gpg密钥，如失败多重试几次
rpm --import 'https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc'
rpm --import 'https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.e495bb49cc4bbe5b.key'
rpm --import 'https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9f4587f226208342.key'

# 下载 rpm 包
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.13.0/rabbitmq-server-3.13.0-1.el8.noarch.rpm

# 安装
rpm -ivh rabbitmq-server-3.13.0-1.el8.noarch.rpm

进行配置

# 启用管理界面插件
rabbitmq-plugins enable rabbitmq_management

# 启动 rabbitmq 服务：
systemctl start rabbitmq-server

# 将 rabbitmq 服务设置为开机自动启动
systemctl enable rabbitmq-server

# 新增登录账号密码 - 用户名 密码
rabbitmqctl add_user yan 123456

# 设置登录账号权限，该用户设置角色是administrator管理员。再设置权限
rabbitmqctl set_user_tags yan administrator
rabbitmqctl set_permissions -p / yan ".*" ".*" ".*"

# 查询当前用户和角色
rabbitmqctl list_users

# 配置所有稳定功能 flag 启用
rabbitmqctl enable_feature_flag all

# 重启rabbitmq服务生效
systemctl restart rabbitmq-server

# 启动rabbitmq服务器
systemctl start rabbitmq-server

# 查看rabbitmq服务器状态
systemctl status rabbitmq-server

# 停止rabbitmq服务器
systemctl stop rabbitmq-server

# 重启rabbitmq服务器
systemctl restart rabbitmq-server

首尾配置

# 删除yum库配置文件，避免下次yum安装在执行一遍
rm -rf /etc/yum.repos.d/rabbitmq.repo

# 分别在不同服务器修改主机名称
vim /etc/hostname
# 分别起名 node01、node02、node03
# 例子：机器1，该文件/etc/hostname 下 填写node1
#      机器2，该文件/etc/hostname 下 填写node2
#      机器3，该文件/etc/hostname 下 填写node3

集群配置

机器1配置：

  1. 分别在3台机器设置ip 地址到对应主机名称的映射

    修改文件/etc/hosts，追加如下内容：

    ip1 node01
ip2 node02
ip3 node03

  2. 查看该mq的cookie值并记录

    cat /var/lib/rabbitmq/.erlang.cookie
# 并记录该值，后续机器2,机器3上的cookie值保持一致
# 在node01节点运行，把该文件复制到其他机器上 
scp /var/lib/rabbitmq/.erlang.cookie root@node02:/var/lib/rabbitmq/.erlang.cookie
scp /var/lib/rabbitmq/.erlang.cookie root@node03:/var/lib/rabbitmq/.erlang.cookie
# 需要将rabbitmq分别重启
systemctl stop rabbitmq-server
systemctl start rabbitmq-server

  3. 重置节点应用

    rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app

机器2配置：

  1. 分别在3台机器设置ip 地址到对应主机名称的映射

    修改文件/etc/hosts，追加如下内容：

    ip1 node01
ip2 node02
ip3 node03

  2. 查看该mq的cookie值并记录

    cat /var/lib/rabbitmq/.erlang.cookie
# 把机器1上的cookie值复制到这里，与机器1保持一致

  3. 重置节点应用，并加入集群

    rabbitmqctl stop_app
rabbitmqctl reset
# 加入集群，去node01机器上找rabbit的程序，进行加入
rabbitmqctl join_cluster rabbit@node01
rabbitmqctl start_app

机器3配置：

  1. 分别在3台机器设置ip 地址到对应主机名称的映射

    修改文件/etc/hosts，追加如下内容：

    ip1 node01
ip2 node02
ip3 node03

  2. 查看该mq的cookie值并记录

    cat /var/lib/rabbitmq/.erlang.cookie
# 把机器1上的cookie值复制到这里，与机器1保持一致

  3. 重置节点应用，并加入集群

    rabbitmqctl stop_app
rabbitmqctl reset
# 加入集群，去node01机器上找rabbit的程序，进行加入
rabbitmqctl join_cluster rabbit@node01
rabbitmqctl start_app

  4. 查看集群状态

    rabbitmqctl cluster_status
# 也可以在图形化界面上，查看集群节点，在nodes这里查看集群节点

负载均衡

一般mq的服务端口5672、ui管理界面端口15672

浏览器访问，某个端口，通过haproxy工具，进行分流每个节点的15672端口上，进行管理ui上的负载均衡；

服务端访问，某个端口，通过haproxy工具，进行分流每个节点的5672端口上，进行服务端的负载均衡；

安装haproxy

yum install -y haproxy
# 查看安装的版本
haproxy -v
# 启动该程序
systemctl start haproxy
# 设置该服务开机自动启动
systemctl enable haproxy
# 查看状态
systemctl status haproxy
# 停止启动
systemctl stop haproxy
# 重启
systemctl restart haproxy
# 查看当前的线程
ps -ef|grep haproxy

修改配置文件

  1. 修改配置文件

    vim /etc/haproxy/haproxy.cfg

    文件内容：

    # 在最后一行添加该内容，该部分配置ui管理器上的负载均衡
# 起mq的ui的名字（rabbitmq_ui_frontend），前端部分配置
frontend rabbitmq_ui_frontend
# 绑定一个地址，端口号，外部浏览器访问该ip地址和端口后，然后通过haproxy进行转发
bind 192.168.xxx.xxx:22222
# http协议
mode http
# 默认的后端服务名，使用下面的名字
default_backend rabbitmq_ui_backend

# 起mq的服务端的名字（rabbitmq_ui_backend）
backend rabbitmq_ui_backend
# http协议
mode http
# 启用轮询的负载均衡
balance roundrobin
option httpchk get /
# 分别设置，不同的mq服务节点的ip地址
server rabbitmq_ui1 192.168.xxx.1:15672 check
server rabbitmq_ui2 192.168.xxx.2:15672 check
server rabbitmq_ui3 192.168.xxx.3:15672 check

# 配置服务端（生产者、消费者）的负载均衡
# 服务端（生产者、消费者）的名字
frontend rabbitmq_frontend
# 绑定一个地址，端口号，服务端（生产者、消费者）访问该ip地址和端口后，然后通过haproxy进行转发
bind 192.168.xxx.xxx:11111
# tcp协议
mode tcp
default_backend rabbitmq_backend
# 与上面的名字保持一致
backend rabbitmq_backend
# tcp协议
mode tcp
# 启用轮询的负载均衡
balance roundrobin
# 分别设置，不同的mq服务节点的ip地址
server rabbitmq1 192.168.xxx.1:5672 check
server rabbitmq2 192.168.xxx.2:5672 check
server rabbitmq3 192.168.xxx.3:5672 check

  2. 设置selinux策略，允许haproxy拥有权限连接任意端口：

    # 保存上面的配置文件，执行下面
setsebool -p haproxy_connect_any=1

# 绑定非本机的ip需要在sysctl.conf文件中配置：
vi   /etc/sysctl.conf
# 添加
net.ipv4.ip_nonlocal_bind=1
# 保存退出，执行
sysctl -p

    selinux是linux系统中的安全模块，它可以限制进程的权限以提高系统的安全性。在某些情况下，selinux可能会阻止haproxy绑定指定的端口，这就需要通过设置域（domain）的安全策略来解决此问题。通过执行setsebool -p haproxy_connect_any=1命令，您已经为haproxy设置了一个布尔值，允许haproxy连接到任意端口。这样，haproxy就可以成功绑定指定的socket，并正常工作。

  3. 重启haproxy

    # 测试配置文件的正确性
haproxy -c -f /etc/haproxy/haproxy.cfg
# 进行重启
systemctl restart haproxy

配置成功

# 访问mq管理ui页面：
http://192.168.xxx.xxx:22222
# 生产者、消费者服务连接mq
  rabbitmq:
    host: 192.168.xxx.xxx
    port: 11111
    username: guest
    password: 123456
# 可通过生产者发送消息，监听消费者是否接受到消息

仲裁队列

仲裁队列是3.8版本以后才有的新功能，是主从模式，支持主从数据同步，在添加队列，队列类型为quorum就是仲裁队列。

创建交换机

与之前创建普通交换机一致

创建仲裁队列
在这里插入图片描述

绑定交换机

与之前的一致

验证，停止某个节点

# 停止rabbit应用
rabbitmqctl stop_app
# 我们在某个节点进行停止mq，然后生产者发送消息，查看消费者是否可以接受到，判断集群的数据一致性
# 默认，主节点挂掉，默认重新选举新的主节点

流式队列

流式队列是3.9版本以后才有的新功能，每个消息都分配一个偏移量，该消息被消费掉也不会被删除，可以重复消费；类似kafka，但没有达到kafka的效果。

前提准备 - 安装相关插件

# 启用stream插件
rabbitmq-plugins enable rabbitmq_stream

# 重启rabbit应用
rabbitmqctl stop_app
rabbitmqctl start_app

# 查看插件状态
rabbitmq-plugins list

负载均衡配置

frontend rabbitmq_stream_frontend
bind 192.168.xxx.100:33333
mode tcp
default_backend rabbitmq_stream_backend

backend rabbitmq_stream_backend
mode tcp
balance roundrobin
# 它的端口是5552
server rabbitmq1 192.168.xxx.1:5552 check
server rabbitmq2 192.168.xxx.1:5552 check
server rabbitmq3 192.168.xxx.2:5552 check

java相关代码

// 引包
<dependencies>
    <dependency>
        <groupid>com.rabbitmq</groupid>
        <artifactid>stream-client</artifactid>
        <version>0.15.0</version>
    </dependency>

    <dependency>
        <groupid>org.slf4j</groupid>
        <artifactid>slf4j-api</artifactid>
        <version>1.7.30</version>
    </dependency>
    <dependency>
        <groupid>ch.qos.logback</groupid>
        <artifactid>logback-classic</artifactid>
        <version>1.2.3</version>
    </dependency>
</dependencies>

创建stream，不需要创建交换机

// 代码创建
environment environment = environment.builder()
        .host("192.168.xxx.1")
        .port(33333)
        .username("用户名")
        .password("密码")
        .build();

environment.streamcreator().stream("stream.atguigu.test2").create();
environment.close();

通过管理ui页面创建的话，队列类型选择：stream

生产者

environment environment = environment.builder()
        .host("192.168.xxx.1")
        .port(33333)
        .username("用户名")
        .password("密码")
        .build();

producer producer = environment.producerbuilder()
        // 队列名称
        .stream("stream.test")
        .build();
// 消息内容
byte[] messagepayload = "hello rabbit stream".getbytes(standardcharsets.utf_8);

countdownlatch countdownlatch = new countdownlatch(1);
producer.send(
        producer.messagebuilder().adddata(messagepayload).build(),
        confirmationstatus -> {
            if (confirmationstatus.isconfirmed()) {
                system.out.println("[生产者端]the message made it to the broker");
            } else {
                system.out.println("[生产者端]the message did not make it to the broker");
            }
            countdownlatch.countdown();
        });

countdownlatch.await();
producer.close();
environment.close();

消费者

environment environment = environment.builder()
        .host("192.168.xxx.1")
        .port(33333)
        .username("用户名")
        .password("密码")
        .build();
environment.consumerbuilder()
    	// 队列名称
        .stream("stream.test")
    	// 这个是消费端起名字
        .name("stream.test.consumer")
        .autotrackingstrategy()
        .builder()
        .messagehandler((offset, message) -> {
            // 接受消息
            byte[] bodyasbinary = message.getbodyasbinary();
            string messagecontent = new string(bodyasbinary);
            system.out.println("[消费者端]messagecontent = " + messagecontent + " offset=" + offset.offset());
        })
        .build();

指定偏移量消费

// 指定offset消费
environment environment = environment.builder()
        .host("192.168.xxx.1")
        .port(33333)
        .username("用户名")
        .password("密码")
        .build();

countdownlatch countdownlatch = new countdownlatch(1);
consumer consumer = environment.consumerbuilder()
        .stream("stream.test")
        .offset(offsetspecification.first())
        .messagehandler((offset, message) -> {
            byte[] bodyasbinary = message.getbodyasbinary();
            string messagecontent = new string(bodyasbinary);
            system.out.println("[消费者端]messagecontent = " + messagecontent);
            countdownlatch.countdown();
        })
        .build();

countdownlatch.await();
consumer.close();

federation插件

它是mq在不同的broker节点之间进行消息传递而无须建立集群，可以实现跨集群的数据同步。

federation交换机

  1. 前提准备：启用该插件、分为上游和下游，创建2个独立的mq服务；

    # 使用docker进行安装mq1服务
docker run -d \
--name rabbitmq-shenzhen \
-p 51000:5672 \
-p 52000:15672 \
-v rabbitmq-plugin:/plugins \
-e rabbitmq_default_user=guest \
-e rabbitmq_default_pass=123456 \
rabbitmq:3.13-management

# 使用docker进行安装mq2服务
docker run -d \
--name rabbitmq-shanghai \
-p 61000:5672 \
-p 62000:15672 \
-v rabbitmq-plugin:/plugins \
-e rabbitmq_default_user=guest \
-e rabbitmq_default_pass=123456 \
rabbitmq:3.13-management

  2. 启用插件

    # 分别在2台mq服务，进行安装该插件
rabbitmq-plugins enable rabbitmq_federation
rabbitmq-plugins enable rabbitmq_federation_management

    安装完插件，在右侧菜单栏可看到

  3. 添加上游连接端点

    在下游的mq服务，进行操作，点击右侧federation upstreams，进行添加，需要起一个上游的连接点名字，（federation-upstreams），url填写：amqp://用户名:密码@访问ip:端口 （是上游的可访问的地址）

    在这里插入图片描述

  4. 创建控制策略

    进行配置，在下游的mq服务，进行操作

    在这里插入图片描述

  5. 验证

    # 需要满足：
# 普通交换机和联邦交换机名称要一致
# 交换机名称要能够和策略正则表达式匹配上
# 发送消息时，两边使用的路由键也要一致
# 队列名称不要求一致

    在这里插入图片描述
    生产者（上游）发送消息、消息会被上游（mq）的队列接收到，然后下游（mq）的队列也会接受到消息；

    满足不同集群进行数据同步；

federation队列

federation队列和federation交换机的最核心区别就是：

  • federation police作用在交换机上，就是federation交换机
  • federation police作用在队列上，就是federation队列
赞 (0)

相关文章:

版权声明：本文内容由互联网用户贡献，该文观点仅代表作者本人。本站仅提供信息存储服务，不拥有所有权，不承担相关法律责任。 如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 2386932994@qq.com 举报，一经查实将立刻删除。

发表评论

验证码：
最近更新的文章
推荐阅读
大家感兴趣的文章
Copyright © 2017-2025  代码网 保留所有权利. 粤ICP备2024248653号
站长QQ:2386932994 | 联系邮箱：2386932994@qq.com