RMQrabbitmq高可用设计

Posted 2022-07-11 DonotCTR

1、概述

        rabbitmq是一个消息队列中间件，这就不细说，公司业务规模越来越大，对于服务和中间件等高可用需求也是越来越多。之前公司在rabbitmq方面都是单机运行在项目服务器，那么存在一个巨大的安全隐患，当rabbitmq产生大量推送时，相关服务器的CPU利用率过大，导致影响相关项目运行。为了解决此类问题，首先将rabbitmq服务进行集中化管理，独立服务器搭建rabbitmq服务，防止单节点宕机影响服务，设计高可用方式运行。架构如下图：

1. 通过公网域名访问rabbitmqweb管理后台（CLB绑定15672端口）
2. rabbitmq01和rabbitmq02通过镜像方式相互备份数据
3. 程序通过内网CLB访问rabbitmq-5762端口实现消息队列生产和消费

2、基础环境

      服务器基础配置：centos74 8VCPU16G

      rabbitmq相关软件包：

      链接：​​百度网盘​​

      提取码：gEt5

# hostnamectl set-hostname rabbitmq01   #根据实际服务器hostname修改
# vim  /etc/hosts
172.16.64.10  rabbitmq01
172.16.48.10  rabbitmq02
# vim /etc/security/limits.conf   ——————》需要重启服务器，但是腾讯云主机已经优化过，可以不需要优化
* soft noproc 65535
* hard noproc 65535
* hard nofile 65535
* soft nofile 65535

        安装rabbitmq和erlang

# rpm -ivh esl-erlang_19.3-1_centos_7_amd64.rpm  --nodeps
# rpm -ivh rabbitmq-server-3.7.7-1.el7.noarch.rpm  --nodeps

         rabbitmq基础配置

# vim /etc/rabbitmq/rabbitmq-env.conf 
MNESIA_BASE=/data/rabbitmq/mnesiad
LOG_BASE=/data/rabbitmq/logs
NODENAME=rabbit@rabbitmq01   #根据实际服务器hostname填写
# mkdir /data/rabbitmq/mnesiad,logs -p
# chown rabbitmq:rabbitmq /data/rabbitmq -R
# vim /etc/rabbitmq/rabbitmq.config 
[
rabbit, [vm_memory_high_watermark_paging_ratio, 0.6,
         vm_memory_high_watermark, 0.6]
].
# more /var/lib/rabbitmq/.erlang.cookie   #特别注意两台服务器一定要一致，否则无法连接
TOGKDYDQGFWFURZUAUSV

       rabbitmq启动

# rabbitmq-plugins enable rabbitmq_management   #加载插件
# rabbitmq-server -detached   #启动rabbitmq
# systemctl enable rabbitmq-server  #设置为开机自启动
# systemctl stop rabbitmq-server   #关闭

       rabbitmq新增用户和授权

# rabbitmqctl list_users      #查看已有用户
//添加用户，命令格式：rabbitmqctl add_user userName Password
# rabbitmqctl add_user admin Jzkj@2020
//给用户添加权限：userName为用户名， Tag为角色名(对应于administrator，monitoring，policymaker，management，或其他自定义名称):，命令格式：rabbitmqctl set_user_tags userName tags
# rabbitmqctl set_user_tags admin administrator
# rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

       在rabbitmq01上进行集群配置

# rabbitmqctl cluster_status   #查看集群
# rabbitmqctl stop_app
# rabbitmqctl reset
# rabbitmqctl join_cluster rabbit@rabbitmq02 加入集群，名为rabbitmq02
# rabbitmqctl start_app
# rabbitmqctl cluster_status
# rabbitmqctl set_policy admin ".*" "ha-mode":"all","ha-sync-mode":"automatic"  
//启动镜像同步操作，切记一定操作

      调整rabbitmq的socket

# vim /usr/lib/systemd/system/rabbitmq-server.service 
[Service]
LimitNOFILE=1000000
# systemctl daemon-reload

      延时队列插件安装可参考：​​rabbitmq延时功能​​

3、高可用构建

         rabbitmq共有四个端口，但实际上可以用上的端口为5672和15672,5672用于程序访问，15672用来web、控制登录，所以分别配置了两个负载均衡，5672用于内网中的程序，所以采用内网负载均衡器，15672主要外网访问，所以采用外网负载均衡进行访问。

        由于我们这边使用的是腾讯云CVM，所以为了实现高可用，则使用腾讯云的CLB，可参考腾讯云官方文档：​​CLB​​

        登录CLB管理后台，如下图为5672端口，一般情况都是使用四层代理

         15672端口配置，用于web控制台登录访问

          登陆rabbitmq控制台

4、高可用测试

         生产者测试脚本

import json
import datetime
import pika
import sys

def get_message():
    for i in range(1000): 
        message = json.dumps(
            id: "10000%s" % i, "amount": 100 * i, "name": "tony", "createtime": str(datetime.datetime.now()))
        producter(message)

def producter(message):

    credentials = pika.PlainCredentials(admin,*********)
    connection = pika.BlockingConnection(pika.ConnectionParameters( 0.0.0.0,5672,/,credentials))
    channel = connection.channel()

    channel.queue_declare(queue=test)
    for i in range(10000):
        channel.basic_publish(exchange=,
                              routing_key=test,
                              body=message)
        print(message)
    connection.close()
    sys.exit()


if __name__ == __main__:
    get_message()

        消费者测试脚本

import pika

def test():
    credentials = pika.PlainCredentials(admin,***********)
    connection = pika.BlockingConnection(pika.ConnectionParameters(0.0.0.0,5672,/,credentials))
    channel = connection.channel()
    channel.queue_declare(queue=balance)

    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)

    channel.basic_consume(balance, callback, False)
    print( [*] Waiting for messages. To exit press CTRL+C)
    channel.start_consuming()


if __name__ == __main__:
    test()