Linux安装RabbitMQ

1. 作者QQ：67065435 QQ群：669756510

2. 本站内容全部为作者原创，转载请注明出处！

参考链接

1. 参考链接

版本映射

1. 查看版本映射

准备工作

1. 准备三台服务器

   192.168.1.11
   192.168.1.12
   192.168.1.13
   

Erlang安装(三台服务器都执行)

1. 本地下载

2. 下载地址

   # 注意：这是RabbitMQ官方推荐的erlang的轻量版rpm包不是erlang官方提供的
   # 完整rpm包，如果你是erlang开发人员，请前往erlang官方下载完整rpm包。
   cd /root
   
   wget --no-check-certificate -c https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-erlang/gpg.E495BB49CC4BBE5B.key
   rpm --import gpg.E495BB49CC4BBE5B.key
   
   wget --no-check-certificate -c https://github.com/rabbitmq/erlang-rpm/releases/download/v26.0.2/erlang-26.0.2-1.el7.x86_64.rpm
   rpm -ivh erlang-26.0.2-1.el7.x86_64.rpm
   

3. 测试启动erlang

   /usr/bin/erl
   # 显示Erlang/OTP 26
   
   # 输入:
   1>9+7.
   16
   
   # 退出:
   2>halt().
   

RabbitMQ安装(三台服务器都执行)

1. 本地下载

2. 下载地址

   cd /root
   
   wget --no-check-certificate -c https://github.com/rabbitmq/signing-keys/releases/download/3.0/rabbitmq-release-signing-key.asc
   rpm --import rabbitmq-release-signing-key.asc
   
   wget --no-check-certificate -c https://dl.cloudsmith.io/public/rabbitmq/rabbitmq-server/gpg.9F4587F226208342.key
   rpm --import gpg.9F4587F226208342.key
   
   wget --no-check-certificate -c https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.12.1/rabbitmq-server-3.12.1-1.suse.noarch.rpm
   rpm -ivh rabbitmq-server-3.12.1-1.suse.noarch.rpm
   

RabbitMQ配置(三台服务器都执行)

1. 创建目录

   mkdir /data
   mkdir -p /data/rabbitmq
   mkdir -p /data/rabbitmq/log
   mkdir -p /data/rabbitmq/data
   chown -R rabbitmq:rabbitmq /data/rabbitmq
   

2. 创建配置(1)

   # 修改环境配置(注意：‘=’两端不能有空格，否则配置识别异常)
   # 【NODENAME】注释则使用默认配置，【当前主机名】通过‘hostname’命令可获取
   
   vim /etc/rabbitmq/rabbitmq-env.conf
   
   # NODENAME=rabbit@当前主机名
   CONSOLE_LOG=use
   LOG_BASE=/data/rabbitmq/log
   MNESIA_BASE=/data/rabbitmq/data
   CONFIG_FILE=/etc/rabbitmq/rabbitmq.config
   ADVANCED_CONFIG_FILE=/etc/rabbitmq/advanced.config
   
   ESC
   :wq
   

3. 创建配置(2)

   # 服务器192.168.1.11配置(另外两台同理可得)
   vim /etc/rabbitmq/rabbitmq.config
   [
     {
       rabbit,
       [
         {
           tcp_listeners,
           [
             {"192.168.1.11",5678},
             {"127.0.0.1",5678},
             {"::1",5678}
           ]
         },
         %%{
         %%  ssl_listeners,
         %%  [5677]
         %%},
         %%{
         %%  ssl_options,
         %%  [
         %%    {cacertfile,"/xxx/ca_cert.pem"},
         %%    {certfile,"/xxx/client_cert.pem"},
         %%    {keyfile,"/xxx/client_key.pem"},
         %%    {verify,verify_peer},
         %%    {fail_if_no_peer_cert,true}
         %%  ]
         %%},
         {
           vm_memory_high_watermark,
           {relative,0.5}
         },
         {
           vm_memory_high_watermark_paging_ratio,
           0.8
         },
         {
           log_levels,
           [
             {connection,warning},
             {channel,warning}
           ]
         }
       ]
     },
     {
       rabbitmq_management,
       [
         {
           tcp_config,
           [{ip,"192.168.1.11"},{port,15678},{ssl,false}]
         }
       ]
     },
     {
       kernel,
       [
         {inet_dist_listen_min,25678},
         {inet_dist_listen_max,25678}
       ]
     }
   ].
   
   ESC
   :wq
   

4. 创建配置(3)

   vim /etc/rabbitmq/advanced.config
   
   [].
   
   ESC
   :wq
   

RabbitMQ插件(主服务器执行)

1. 安装管理插件

   # 启用管理插件
   rabbitmq-plugins enable rabbitmq_management
   
   # 查看启用插件
   rabbitmq-plugins list
   

RabbitMQ启动(三台服务器都执行)

1. 修改单元文件

   vim /usr/lib/systemd/system/rabbitmq-server.service
   
   [Unit]
   Description=RabbitMQ
   After=syslog.target network.target
   
   [Service]
   Type=notify
   User=rabbitmq
   Group=rabbitmq
   UMask=0027
   NotifyAccess=all
   Restart=on-failure
   WorkingDirectory=/data/rabbitmq
   ExecStart=/usr/sbin/rabbitmq-server
   ExecStop=/usr/sbin/rabbitmqctl shutdown
   SuccessExitStatus=69
   TimeoutStartSec=600
   LimitNOFILE=32768
   PrivateTmp=true
   RestartSec=10
   
   [Install]
   WantedBy=multi-user.target
   
   ESC
   :wq
   

2. 管理服务进程

   systemctl daemon-reload
   systemctl enable rabbitmq-server
   systemctl restart rabbitmq-server
   systemctl status rabbitmq-server
   

RabbitMQ管理(主服务器执行)

1. 添加用户并授权

   # 删除默认vhost
   rabbitmqctl delete_vhost /
   
   # 删除默认用户
   rabbitmqctl delete_user guest
   
   # 创建用户(请将'**************'改成更安全的密码)
   rabbitmqctl add_user rabbitmq '**************'
   
   # 修改密码(如果忘记密码可以这样修改密码)
   rabbitmqctl change_password rabbitmq '**************'
   
   # 添加用户[管理员]标签
   rabbitmqctl set_user_tags rabbitmq administrator
   
   # 创建virtual host(‘/stat’是可以自己命名的)
   rabbitmqctl add_vhost /stat
   
   # 为用户授权可以访问的[virtual host]和操作类型[配置、读、写]
   rabbitmqctl set_permissions -p /stat rabbitmq '.*' '.*' '.*'
   

拷贝cookie(主服务器执行)

1. 拷贝cookie

   scp -r /var/lib/rabbitmq/.erlang.cookie root@192.168.1.12:/var/lib/rabbitmq/.erlang.cookie
   scp -r /var/lib/rabbitmq/.erlang.cookie root@192.168.1.13:/var/lib/rabbitmq/.erlang.cookie
   

RabbitMQ端口(三台服务器都执行)

1. 开放防火墙端口

   firewall-cmd --permanent --zone=public --add-port='4369/tcp'
   firewall-cmd --permanent --zone=public --add-port='5678/tcp'
   firewall-cmd --permanent --zone=public --add-port='15678/tcp'
   firewall-cmd --permanent --zone=public --add-port='25678/tcp'
   firewall-cmd --reload
   

RabbitMQ集群(从服务器执行)

1. 节点加入集群

   rabbitmqctl stop_app
   rabbitmqctl reset
   rabbitmqctl join_cluster rabbit@当前主机名 --ram
   # 修改节点类型
   # rabbitmqctl change_cluster_node_type ram
   rabbitmqctl start_app
   

附录①：RabbitMQ使用(PHP代码服务器执行)

1. 更新CMAKE

2. [安装librabbitmq]

   yum -y --nogpgcheck install librabbitmq
   

3. 安装rabbitmq-c

   cd /root
   wget --no-check-certificate -c https://github.com/alanxz/rabbitmq-c/archive/refs/tags/v0.13.0.tar.gz -O rabbitmq-c-0.13.0.tar.gz
   tar -xf rabbitmq-c-0.13.0.tar.gz
   cd rabbitmq-c-0.13.0
   
   # ↓下面↓命令中的‘gcc’、‘g++’请根据自己的系统灵活变通
   cmake /root/rabbitmq-c-0.13.0 \
   -DCMAKE_INSTALL_PREFIX=/usr \
   -DCMAKE_C_COMPILER=/usr/bin/gcc \
   -DCMAKE_CXX_COMPILER=/usr/bin/g++
   
   make -j$(nproc)
   make -j$(nproc) install
   

4. 安装php-amqp ``` cd /root wget --no-check-certificate -c https://github.com/php-amqp/php-amqp/archive/refs/tags/v1.11.0.tar.gz -O php-amqp-1.11.0.tar.gz tar -xf php-amqp-1.11.0.tar.gz cd php-amqp-1.11.0 phpize ./configure make -j$(nproc) make -j$(nproc) install

vim /usr/local/php/lib/php.ini

extension=amqp.so

ESC :wq

1. php-amqp生产者
   ```php
   <?php
   /**
    * $prefetch_count 用来设定手动ACK时，消费者一次获取到消息的数
    * 量，该参数用来保证客户端总是处于无缝衔接的状态，参考链接：
    * https://blog.rabbitmq.com/posts/2012/05/some-queuing-theory-throughput-latency-and-bandwidth
    * (下发时间 + 应答时间 + 处理时间)/处理时间
    *
    * FANOUT 交换机将消息发送到【$exchange bind $queue】的所有队列
    * DIRECT 交换机将消息发送到【$binding_key == $routing_key】的所有队列
    * TOPIC  交换机将消息发送到【$binding_key like $routing_key】的所有队列
    * HEADERS交换机将消息发送到【$arguments['xxx'] == $attributes['xxx']】的所有队列(性能差)
    *
    * TOPIC模式的通配符只有"*"、"#"两种，且通配符前面必须加上"."
    * binding_key匹配规则
    * ‘*’匹配  1  个单词: rabbit.* 能匹配 rabbit.a、rabbit.b、rabbit.c 等
    * ‘#’匹配 0-∞ 个单词: rabbit.# 能匹配 rabbit、rabbit.a、rabbit.a.b 等
    *
    * HEADERS模式的匹配规则有两种
    * x-match==any
    * $arguments = ['x-match' => 'any', 'user_nick' => '张三', 'user_age' => 18];可以匹配
    * $attributes = ['user_nick' => '张三', 'user_age' => 18];
    * $attributes = ['user_nick' => '李四', 'user_age' => 18];
    * x-match==all
    * $arguments = ['x-match' => 'all', 'user_nick' => '张三', 'user_age' => 18];只能匹配
    * $attributes = ['user_nick' => '张三', 'user_age' => 18];
    */
   try {
       $connect = new \AMQPConnection([
           "host" => '127.0.0.1',
           "port" => 5678,
           "login" => 'rabbitmq',
           "password" => '**************',
           "vhost" => '/stat',
       ]);
       $prefetch_count = 10;
       $connect->connect();
       $channel = new \AMQPChannel($connect);
       $channel->setPrefetchCount($prefetch_count);

       $exchange_name = "exchange.a";
       $exchange = new \AMQPExchange($channel);
       $exchange->setName($exchange_name);
       $exchange->setFlags(AMQP_DURABLE);//交换机持久化
       $exchange->setType(AMQP_EX_TYPE_DIRECT);
       $exchange->declareExchange();

       $queue_name = "queue.a";
       $queue = new \AMQPQueue($channel);
       $queue->setName($queue_name);
       $queue->setFlags(AMQP_DURABLE);//队列持久化
       $queue->declareQueue();

       $binding_key = "route.a";
       $arguments = [];
       $queue->bind($exchange_name, $binding_key, $arguments);
       $message = microtime(true);
       $routing_key = "route.a";
       $attributes = [
           'delivery_mode' => AMQP_DURABLE,//消息持久化
       ];
       $exchange->publish($message, $routing_key, AMQP_NOPARAM, $attributes);

       $connect->disconnect();
   } catch (\Exception $e) {
       echo $e->getMessage();
   }

1. php-amqp消费者

   <?php
   try {
       $connect = new \AMQPConnection([
           "host" => '127.0.0.1',
           "port" => 5678,
           "login" => 'rabbitmq',
           "password" => '**************',
           "vhost" => '/stat',
       ]);
       $prefetch_count = 10;
       $connect->connect();
       $channel = new \AMQPChannel($connect);
       $channel->setPrefetchCount($prefetch_count);
   
       $queue_name = "queue.a";
       $queue = new \AMQPQueue($channel);
       $queue->setName($queue_name);
       $queue->setFlags(AMQP_DURABLE);//队列持久化
       $queue->declareQueue();
   
       while (true) {
           //不能保证绝对处理成功，所以手动ACK应答
           $queue->consume(function ($envelope, $queue) {
               $message = $envelope->getBody();
               //假设 1-成功 2-失败
               if (rand(1, 2) === 1) {
                   echo $message, "成功", PHP_EOL;
                   $queue->ack($envelope->getDeliveryTag());
               } else {
                   echo $message, "失败", PHP_EOL;
                   $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
               }
           });
   
           //可以保证绝对处理成功，所以自动ACK应答
           //$queue->consume(function ($envelope, $queue) {
           //    $message = $envelope->getBody();
           //    echo $message, "成功", PHP_EOL;
           //}, AMQP_AUTOACK);
       }
       $connect->disconnect();
   } catch (\Exception $e) {
       echo $e->getMessage();
   }
   

2. php-amqp生产者(定时版)

   <?php
   try {
       $connect = new \AMQPConnection([
           "host" => '127.0.0.1',
           "port" => 5678,
           "login" => 'rabbitmq',
           "password" => '**************',
           "vhost" => '/stat',
       ]);
       $prefetch_count = 10;
       $connect->connect();
       $channel = new \AMQPChannel($connect);
       $channel->setPrefetchCount($prefetch_count);
   
       $exchange_name = "exchange.a";
       $exchange = new \AMQPExchange($channel);
       $exchange->setName($exchange_name);
       $exchange->setFlags(AMQP_DURABLE);//交换机持久化
       $exchange->setType('x-delayed-message');            //消息定时关键代码
       $exchange->setArgument('x-delayed-type', 'direct'); //消息定时关键代码
       $exchange->declareExchange();
   
       $queue_name = "queue.a";
       $queue = new \AMQPQueue($channel);
       $queue->setName($queue_name);
       $queue->setFlags(AMQP_DURABLE);//队列持久化
       $queue->declareQueue();
   
       $binding_key = "route.a";
       $arguments = [];
       $queue->bind($exchange_name, $binding_key, $arguments);
       $message = microtime(true);
       $routing_key = "route.a";
       $ttl = 10;//10秒后发消息                                                 //消息定时关键代码
       $attributes = [                                                         //消息定时关键代码
           'delivery_mode' => AMQP_DURABLE,//消息持久化                         //消息定时关键代码
           'headers' => ['x-delay' => $ttl * 1000],                            //消息定时关键代码
       ];                                                                      //消息定时关键代码
       $exchange->publish($message, $routing_key, AMQP_NOPARAM, $attributes);  //消息定时关键代码
   
       $connect->disconnect();
   } catch (\Exception $e) {
       echo $e->getMessage();
   }
   

3. php-amqp消费者(定时版)

   <?php
   try {
       $connect = new \AMQPConnection([
           "host" => '127.0.0.1',
           "port" => 5678,
           "login" => 'rabbitmq',
           "password" => '**************',
           "vhost" => '/stat',
       ]);
       $prefetch_count = 10;
       $connect->connect();
       $channel = new \AMQPChannel($connect);
       $channel->setPrefetchCount($prefetch_count);
   
       $queue_name = "queue.a";
       $queue = new \AMQPQueue($channel);
       $queue->setName($queue_name);
       $queue->setFlags(AMQP_DURABLE);//队列持久化
       $queue->declareQueue();
   
       while (true) {
           //不能保证绝对处理成功，所以手动ACK应答
           $queue->consume(function ($envelope, $queue) {
               $message = $envelope->getBody();
               //假设 1-成功 2-失败
               if (rand(1, 2) === 1) {
                   echo $message, "成功", PHP_EOL;
                   $queue->ack($envelope->getDeliveryTag());
               } else {
                   echo $message, "失败", PHP_EOL;
                   $queue->nack($envelope->getDeliveryTag(), AMQP_REQUEUE);
               }
           });
   
           //可以保证绝对处理成功，所以自动ACK应答
           //$queue->consume(function ($envelope, $queue) {
           //    $message = $envelope->getBody();
           //    echo $message, "成功", PHP_EOL;
           //}, AMQP_AUTOACK);
       }
       $connect->disconnect();
   } catch (\Exception $e) {
       echo $e->getMessage();
   }
   

附录②：RabbitMQ命令(常用)

1. RabbitMQ角色权限

   # 角色列表:
   none、management、policymaker、monitoring、administrator
   
   # 角色权限:
   【none】
   不能访问management_plugin(图形管理界面)
   
   【management】(单一virtual hosts用户)
   none用户可以通过AMQP做的任何事外加：
   列出自己可以通过AMQP登入的virtual hosts  
   查看自己的virtual hosts中的queues, exchanges 和 bindings
   查看和关闭自己的channels 和 connections
   查看有关自己的virtual hosts的“全局”的统计信息，包含其他用户在这些virtual hosts中的活动
   
   【policymaker】
   management用户可以做的任何事外加：
   查看、创建和删除自己的virtual hosts所属的policies和parameters
   
   【monitoring】(监控用户)
   management用户可以做的任何事外加：
   列出所有virtual hosts，包括他们不能登录的virtual hosts
   查看其他用户的connections和channels
   查看节点级别的数据如clustering和memory使用情况
   查看真正的关于所有virtual hosts的全局的统计信息
   
   【administrator】(全权用户)
   policymaker和monitoring用户可以做的任何事外加:
   创建和删除virtual hosts
   查看、创建和删除users
   查看创建和删除permissions
   关闭其他用户的connections
   

2. rabbitmqctl管理方式

   # 启用rabbitmq_management
   rabbitmqctl start_app
   rabbitmq-plugins enable rabbitmq_management
   
   # Web管理：
   http://[ip]:15678
   
   # Api管理：
   http://[ip]:15678/api
   
   # Cli管理：
   http://[ip]:15678/cli
   

3. rabbitmqctl管理【用户】

   # 添加用户
   rabbitmqctl add_user         <username> <password>
   
   # 删除用户
   rabbitmqctl delete_user      <username>
   
   # 修改用户密码
   rabbitmqctl change_password  <username> <newpassword>
   
   # 清除用户密码
   rabbitmqctl clear_password   <username>
   
   # 设置用户标签
   rabbitmqctl set_user_tags    <username> <tag1> <tag2> ... <tag_n>
   
   # 打印用户列表
   rabbitmqctl list_users
   

4. rabbitmqctl管理【vhosts】

   # 创建vhosts
   rabbitmqctl add_vhost    <vhostpath>
   
   # 删除vhosts
   rabbitmqctl delete_vhost <vhostpath>
   
   # 列出vhosts
   rabbitmqctl list_vhosts  <vhostinfoitem1> <vhostinfoitem2> ... <vhostinfoitem_n>
   

5. rabbitmqctl管理【用户权限】

   # 修改权限
   rabbitmqctl set_permissions      [-p <vhostpath>] <user> <conf> <write> <read>
   
   # 删除权限
   rabbitmqctl clear_permissions    [-p <vhostpath>] <username>
   
   # 列出权限
   rabbitmqctl list_permissions     [-p <vhostpath>]
   
   # 列出权限(指定用户)
   rabbitmqctl list_user_permissions <username>
   

6. rabbitmq-plugins命令

   # 打印插件列表 -v:显示插件版本 -m:显示插件名称 -E:显示明确已经开启的插件 -e:显示明确和后台开启的插件
   rabbitmq-plugins list
   
   # 启用插件
   rabbitmq-plugins enable <plugin1> <plugin2> ... <plugin_n>
   
   # 禁用插件
   rabbitmq-plugins disable <plugin1> <plugin2> ... <plugin_n>
   
   # 启用插件(示例)
   rabbitmq-plugins enable amqp_client cowboy cowlib
   

7. rabbitmqctl管理【进程服务】

   # 停止运行rabbitmq
   rabbitmqctl stop
   
   # 停止运行rabbitmq上的应用
   rabbitmqctl stop_app
   
   # 初始化node状态
   rabbitmqctl stop_app
   rabbitmqctl reset
   
   # 强制初始化node状态
   rabbitmqctl stop_app
   rabbitmqctl force_reset
   

8. rabbitmqctl管理【运行环境】

   # 显示broker的状态
   rabbitmqctl status
   
   # 显示环境参数的信息
   rabbitmqctl environment
   

9. rabbitmqctl管理【vhosts参数】

   # 修改vhost路径参数
   rabbitmqctl set_parameter    [-p <vhostpath>] <component_name> <key> <value>
   
   # 删除vhost路径参数
   rabbitmqctl clear_parameter  [-p <vhostpath>] <component_name> <key>
   
   # 列出vhost路径参数
   rabbitmqctl list_parameters  [-p <vhostpath>]
   

10. rabbitmqctl管理【vhosts策略】

    # name:          策略名称
    # pattern:       正则表达式
    # apply-to:      应用范围(all、queues、exchange)
    # priority:      优先级
    # definition:    当前策略
    
    # 设置策略
    rabbitmqctl set_policy       [-p <vhostpath>] [--priority <priority>] [--apply-to <apply-to>] <name> <pattern> <definition(json)>
    
    # 删除策略
    rabbitmqctl clear_policy     [-p <vhostpath>] <name>
    
    # 列出已有的策略
    rabbitmqctl list_policies    [-p <vhostpath>]
    

11. rabbitmqctl打印各种信息

    # 打印队列列表
    rabbitmqctl list_queues      [-p <vhostpath>] [<queueinfoitem> ...]
    
    # 打印交换机列表
    rabbitmqctl list_exchanges   [-p <vhostpath>] [<exchangeinfoitem> ...]
    
    # 打印绑定列表
    rabbitmqctl list_bindings    [-p <vhostpath>] [<bindinginfoitem> ...]
    
    # 打印消费者列表
    rabbitmqctl list_consumers   [-p <vhostpath>]
    
    # 打印连接列表
    rabbitmqctl list_connections [<connectioninfoitem> ...]
    
    # 打印渠道列表
    rabbitmqctl list_channels    [<channelinfoitem> ...]
    

附录③：仅供了解

1. 安装定时插件

   # 仅用来演示RabbitMQ三方插件的安装方式，不要真的安装，这个插件有些坑，具体表现为
   # 1. 当某一队列消费者只有一个时，该队列中的前一个定时消息会阻塞后一个；
   # 2. 当构建高可用集群时，将启用了本插件的节点加入集群会导致启动异常；
   
   # 参考链接
   # https://github.com/rabbitmq/rabbitmq-delayed-message-exchange
   
   # 获取插件目录
   rabbitmq-plugins directories -s
   rabbitmq-plugins directories -s|grep -oE '/usr([^:]*?)server(.*?)plugins'
   
   # 进入插件目录
   cd <上一步获取的目录>
   
   # 下载定时插件
   wget --no-check-certificate -c https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.12.0/rabbitmq_delayed_message_exchange-3.12.0.ez
   
   # 启用定时插件
   rabbitmq-plugins enable rabbitmq_delayed_message_exchange
   
   # 查看启用插件
   rabbitmq-plugins list
   

2. RabbitMQ运行时参数修改

   # 仅供了解RabbitMQ运行时参数修改方法，无需执行
   rabbitmqctl eval 'application:get_env(rabbit,mnesia_table_loading_retry_limit).'
   rabbitmqctl eval 'application:set_env(rabbit,mnesia_table_loading_retry_limit,15).'
   rabbitmqctl eval 'application:get_env(rabbit,mnesia_table_loading_retry_timeout).'
   rabbitmqctl eval 'application:set_env(rabbit,mnesia_table_loading_retry_timeout,60000).'