RabbitMQ（五）集群配置、Management UI

一、安装RabbitMQ

1、前置要求

CentOS发行版的版本CentOS 8 Stream
下载：
官网：https://centos.org/download/
官网下载：https://wiki.centos.org/Download.html
阿里云镜像：http://mirrors.aliyun.com/centos/8-stream/isos/x86_64/
镜像下载：http://mirrors.aliyun.com/centos/8-stream/isos/x86_64/CentOS-Stream-8-x86_64-latest-boot.iso

2、安装docker版

拉取镜像

docker pull rabbitmq:3.13-management

启动容器

# 开三台修改端口地址 麻了，在这弄了好久，node3节点会去找node1和node2，所以需要两个别名都加上，注意hostname和node名称要一致（即下面name和hostname要一致）
docker run -d \
--name rabbitmq-node1 \
--hostname rabbitmq-node1 \
-e RABBITMQ_ERLANG_COOKIE='rabbitcookie' \
-p 5672:5672 \
-p 15672:15672 \
-v rabbitmq-plugin:/plugins \
-e RABBITMQ_DEFAULT_USER=guest \
-e RABBITMQ_DEFAULT_PASS=123456 \
rabbitmq:3.13-management

docker run -d \
--name rabbitmq-node2 \
--hostname rabbitmq-node2 \
-e RABBITMQ_ERLANG_COOKIE='rabbitcookie' \
-p 5673:5672 \
-p 15673:15672 \
--link rabbitmq-node1:rabbitmq-node1 \
-v rabbitmq-plugin:/plugins \
-e RABBITMQ_DEFAULT_USER=guest \
-e RABBITMQ_DEFAULT_PASS=123456 \
rabbitmq:3.13-management

docker run -d \
--name rabbitmq-node3 \
--hostname rabbitmq-node3 \
-e RABBITMQ_ERLANG_COOKIE='rabbitcookie' \
-p 5674:5672 \
-p 15674:15672 \
--link rabbitmq-node1:rabbitmq-node1 \
--link rabbitmq-node2:rabbitmq-node2 \
-v rabbitmq-plugin:/plugins \
-e RABBITMQ_DEFAULT_USER=guest \
-e RABBITMQ_DEFAULT_PASS=123456 \
rabbitmq:3.13-management

使用docker ps

检查是否成功启动

查看所有容器网络配置：默认网络配置为桥接模式

docker inspect --format='{{.Name}} - {{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' $(docker ps -aq)

复制第一个节点的.erlang.cookie

# 先复制到本地/opt/test，不可直接在两个容器中复制会报错
docker cp rabbitmq-node1:/var/lib/rabbitmq/.erlang.cookie /opt/test

docker cp /opt/test/.erlang.cookie  rabbitmq-node2:/var/lib/rabbitmq/.erlang.cookie
docker cp /opt/test/.erlang.cookie  rabbitmq-node3:/var/lib/rabbitmq/.erlang.cookie

在rabbitmq-node1、rabbitmq-node2、rabbitmq-node3查看内容是否一致

more /var/lib/rabbitmq/.erlang.cookie

进入各节点命令行

docker exec -it rabbitmq-node1 /bin/bash
docker exec -it rabbitmq-node2 /bin/bash
docker exec -it rabbitmq-node3 /bin/bash

# 进入rabbitmq-node2容器命令行
apt update
apt install curl -y
curl rabbitmq-node1:15672

检查别名是否生效（因为在容器中命令行无法修改hosts文件，所以在启动时通过配置别名来访问另一个容器），使用curl查看是否有返回

配置集群

进入rabbitmq-node2、rabbitmq-node3容器命令行

# 执行代码
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@rabbitmq-node1
rabbitmqctl start_app

检查集群状态

rabbitmqctl cluster_status

其他博客参考案例：https://blog.csdn.net/want_you_gogo/article/details/120850605

3、三台组合集群安装版

rabbitmq节点

查看rabbitmq节点cookie值并记录

more /var/lib/rabbitmq/.erlang.cookie 

AJRZLSANMKBYKOAXMSIA

重置节点应用

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app

配置hostname

# 修改文件vim /etc/hosts，追加如下内容：
172.17.0.2 node1
172.17.0.3 node2
172.17.0.4 node3

# 退出rabbitmq节点容器
exit

rabbitmq-node2节点

修改rabbitmq节点cookie值和第一个节点一样

vim /var/lib/rabbitmq/.erlang.cookie
# 如果无法修改
# 先退出vim
:!e
:q
# 赋权
chmod 700 /var/lib/rabbitmq/.erlang.cookie

AJRZLSANMKBYKOAXMSIA

配置hostname

# 修改文件vim /etc/hosts，追加如下内容：
172.17.0.2 node1
172.17.0.3 node2
172.17.0.4 node3

重置节点应用（这里注意和第一个节点不同）

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# 退出rabbitmq节点容器
exit

rabbitmq-node3节点

修改rabbitmq节点cookie值和第一个节点一样

vim /var/lib/rabbitmq/.erlang.cookie
# 如果无法修改
# 先退出vim
:!e
:q
# 赋权
chmod 700 /var/lib/rabbitmq/.erlang.cookie

AJRZLSANMKBYKOAXMSIA

配置hostname

# 修改文件vim /etc/hosts，追加如下内容：
172.17.0.2 node1
172.17.0.3 node2
172.17.0.4 node3

重置节点应用

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# 退出rabbitmq节点容器
exit

二、负载均衡：Management UI

1、说明

• 其实访问任何一个RabbitMQ实例的管理界面都是对集群操作，所以配置负载均衡通过统一入口访问在我们学习期间就是锦上添花
• 先给管理界面做负载均衡，然后方便我们在管理界面上创建交换机、队列等操作

2、安装HAProxy

yum install -y haproxy
haproxy -v
systemctl start haproxy
systemctl enable haproxy

3、修改配置文件

配置文件位置：

/etc/haproxy/haproxy.cfg

在配置文件末尾增加如下内容：

frontend rabbitmq_ui_frontend
bind 192.168.200.100:22222
mode http
default_backend rabbitmq_ui_backend

backend rabbitmq_ui_backend
mode http
balance roundrobin
option httpchk GET /
server rabbitmq_ui1 192.168.200.100:15672 check
server rabbitmq_ui2 192.168.200.150:15672 check
server rabbitmq_ui3 192.168.200.200:15672 check

设置SELinux策略，允许HAProxy拥有权限连接任意端口：

setsebool -P haproxy_connect_any=1

SELinux是Linux系统中的安全模块，它可以限制进程的权限以提高系统的安全性。在某些情况下，SELinux可能会阻止HAProxy绑定指定的端口，这就需要通过设置域（domain）的安全策略来解决此问题。

通过执行setsebool -P haproxy_connect_any=1命令，您已经为HAProxy设置了一个布尔值，允许HAProxy连接到任意端口。这样，HAProxy就可以成功绑定指定的socket，并正常工作。

重启HAProxy：

systemctl restart haproxy

检查是否成功组建集群

三、负载均衡：核心功能

新增

vim /etc/haproxy/haproxy.cfg

在配置文件末尾增加如下内容：

frontend rabbitmq_frontend
bind 192.168.217.134:11111
mode tcp
default_backend rabbitmq_backend

backend rabbitmq_backend
mode tcp
balance roundrobin
server rabbitmq1 127.0.0.1:5672 check
server rabbitmq2 127.0.0.1:5673 check
server rabbitmq3 127.0.0.1:5674 check

四、测试

创建组件

• 交换机：exchange.cluster.test
• 队列：queue.cluster.test
• 路由键：routing.key.cluster.test

pom

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.1.5</version>
</parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

生产者

主启动类

package com.atguigu.mq;  
  
import org.springframework.boot.SpringApplication;  
import org.springframework.boot.autoconfigure.SpringBootApplication;  
  
@SpringBootApplication
public class RabbitMQProducerMainType {

    public static void main(String[] args) {
        SpringApplication.run(RabbitMQProducerMainType.class, args);
    }

}

配置YAML

spring:
  rabbitmq:
    host: 192.168.217.134
    port: 11111
    username: guest
    password: 123456
    virtual-host: /
    publisher-confirm-type: CORRELATED # 交换机的确认
    publisher-returns: true # 队列的确认
logging:
  level:
    com.atguigu.mq.config.MQProducerAckConfig: info

配置类

package com.atguigu.mq.config;

import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

@Configuration
@Slf4j
public class MQProducerAckConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("消息发送到交换机成功！数据：" + correlationData);
        } else {
            log.info("消息发送到交换机失败！数据：" + correlationData + " 原因：" + cause);
        }
    }

    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.info("消息主体: " + new String(returned.getMessage().getBody()));
        log.info("应答码: " + returned.getReplyCode());
        log.info("描述：" + returned.getReplyText());
        log.info("消息使用的交换器 exchange : " + returned.getExchange());
        log.info("消息使用的路由键 routing : " + returned.getRoutingKey());
    }
}

测试类

package com.atguigu.mq.test;

import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
public class RabbitMQTest {

    @Resource
    private RabbitTemplate rabbitTemplate;

    public static final String EXCHANGE_CLUSTER_TEST = "exchange.cluster.test";
    public static final String ROUTING_KEY_CLUSTER_TEST = "routing.key.cluster.test";

    @Test
    public void testSendMessage() {
        rabbitTemplate.convertAndSend(EXCHANGE_CLUSTER_TEST, ROUTING_KEY_CLUSTER_TEST, "message test cluster~~~");
    }

}

消费者

主启动类

package com.atguigu.mq;  
  
import org.springframework.boot.SpringApplication;  
import org.springframework.boot.autoconfigure.SpringBootApplication;  
  
@SpringBootApplication
public class RabbitMQProducerMainType {

    public static void main(String[] args) {
        SpringApplication.run(RabbitMQProducerMainType.class, args);
    }

}

配置YAML

spring:
  rabbitmq:
    host: 192.168.217.134
    port: 11111
    username: guest
    password: 123456
    virtual-host: /
    listener:
      simple:
        acknowledge-mode: manual
logging:
  level:
    com.atguigu.mq.listener.MyProcessor: info

监听器

package com.atguigu.mq.listener;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@Slf4j
public class MyProcessor {

    @RabbitListener(queues = {"queue.cluster.test"})
    public void processNormalQueueMessage(String data, Message message, Channel channel) 
        throws IOException {
        
        log.info("消费端：" + data);

        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

}