PHP小码农

Appearance

RabbitMQ rabbitmq-plugins 命令详解 ​

概述 ​

rabbitmq-plugins 是 RabbitMQ 的插件管理命令行工具，用于启用、禁用和管理 RabbitMQ 插件。RabbitMQ 插件系统提供了丰富的扩展功能，包括管理界面、消息协议支持、集群管理、监控集成等。

通过 rabbitmq-plugins 命令，管理员可以灵活地扩展 RabbitMQ 的功能，满足不同业务场景的需求。

核心功能 ​

  • 插件列表查看：列出所有可用和已启用的插件
  • 插件启用：启用指定的插件及其依赖
  • 插件禁用：禁用指定的插件
  • 插件状态检查：查看插件的运行状态
  • 插件目录管理：管理插件安装目录

基本语法 ​

bash
rabbitmq-plugins <command> [command options] [plugin name ...]

常用选项 ​

选项说明
-n node指定节点名称
--offline离线模式，不连接节点
--online在线模式（默认）
-e启用插件（enable 的简写）
-d禁用插件（disable 的简写）

插件管理命令 ​

列出插件 ​

bash
# 列出所有插件
rabbitmq-plugins list

# 列出所有插件（带详细信息）
rabbitmq-plugins list -v

# 列出已启用的插件
rabbitmq-plugins list -e

# 列出所有插件（带状态）
rabbitmq-plugins list -s

输出示例：

Listing plugins with pattern ".*" ...
 Configured: E = explicitly enabled; e = implicitly enabled
 | Status: * = running on rabbit@localhost
 |/
[e*] rabbitmq_amqp1_0                  3.12.0
[E*] rabbitmq_management               3.12.0
[  ] rabbitmq_mqtt                     3.12.0
[e*] rabbitmq_web_dispatch             3.12.0

状态说明：

  • E - 显式启用的插件
  • e - 隐式启用的插件（作为依赖被启用）
  • * - 正在运行
  • 空格 - 未启用

启用插件 ​

bash
# 启用单个插件
rabbitmq-plugins enable <plugin_name>

# 启用多个插件
rabbitmq-plugins enable <plugin1> <plugin2> <plugin3>

# 启用插件（离线模式）
rabbitmq-plugins enable --offline <plugin_name>

示例：

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

# 启用 MQTT 插件
rabbitmq-plugins enable rabbitmq_mqtt

# 启用 STOMP 插件
rabbitmq-plugins enable rabbitmq_stomp

禁用插件 ​

bash
# 禁用单个插件
rabbitmq-plugins disable <plugin_name>

# 禁用多个插件
rabbitmq-plugins disable <plugin1> <plugin2>

# 禁用插件（离线模式）
rabbitmq-plugins disable --offline <plugin_name>

示例：

bash
# 禁用管理界面插件
rabbitmq-plugins disable rabbitmq_management

# 禁用 MQTT 插件
rabbitmq-plugins disable rabbitmq_mqtt

注意：禁用插件时，依赖该插件的其他插件也会被禁用。

查看插件详情 ​

bash
rabbitmq-plugins is_enabled <plugin_name>

示例：

bash
rabbitmq-plugins is_enabled rabbitmq_management

官方插件列表 ​

管理与监控插件 ​

插件名称说明
rabbitmq_managementWeb 管理界面
rabbitmq_management_agent管理代理（管理插件依赖）
rabbitmq_web_dispatchWeb 请求分发
rabbitmq_prometheusPrometheus 监控指标导出
rabbitmq_top查看进程 CPU 使用情况

消息协议插件 ​

插件名称说明
rabbitmq_amqp1_0AMQP 1.0 协议支持
rabbitmq_mqttMQTT 协议支持
rabbitmq_stompSTOMP 协议支持
rabbitmq_web_stompWebSocket STOMP 支持
rabbitmq_web_mqttWebSocket MQTT 支持

集群与高可用插件 ​

插件名称说明
rabbitmq_federation联邦插件（跨集群消息传递）
rabbitmq_federation_management联邦管理界面
rabbitmq_shovel数据迁移工具
rabbitmq_shovel_managementShovel 管理界面

消息处理插件 ​

插件名称说明
rabbitmq_delayed_message_exchange延迟消息交换器
rabbitmq_message_timestamp消息时间戳
rabbitmq_consistent_hash_exchange一致性哈希交换器
rabbitmq_random_exchange随机交换器

安全与认证插件 ​

插件名称说明
rabbitmq_auth_backend_ldapLDAP 认证后端
rabbitmq_auth_backend_httpHTTP 认证后端
rabbitmq_auth_mechanism_sslSSL 客户端证书认证
rabbitmq_auth_backend_cache认证缓存

其他实用插件 ​

插件名称说明
rabbitmq_tracing消息追踪
rabbitmq_event_exchange事件交换器
rabbitmq_jms_topic_exchangeJMS 主题交换器

常用插件详解 ​

rabbitmq_management（管理界面） ​

功能：提供 Web 管理界面和 HTTP API

启用命令：

bash
rabbitmq-plugins enable rabbitmq_management

访问地址：http://localhost:15672

依赖插件：

  • rabbitmq_management_agent
  • rabbitmq_web_dispatch

rabbitmq_prometheus（Prometheus 监控） ​

功能：导出 Prometheus 格式的监控指标

启用命令：

bash
rabbitmq-plugins enable rabbitmq_prometheus

访问地址：http://localhost:15692/metrics

主要指标：

  • 队列消息数
  • 连接数
  • 通道数
  • 消息速率
  • 内存使用

rabbitmq_mqtt（MQTT 协议） ​

功能：支持 MQTT 3.1 和 3.1.1 协议

启用命令：

bash
rabbitmq-plugins enable rabbitmq_mqtt

默认端口：1883（非加密），8883（TLS）

配置示例：

ini
# /etc/rabbitmq/rabbitmq.conf
mqtt.listeners.tcp.default = 1883
mqtt.listeners.ssl.default = 8883
mqtt.default_user = guest
mqtt.default_pass = guest
mqtt.allow_anonymous = false
mqtt.vhost = /
mqtt.exchange = amq.topic
mqtt.max_frame_size = 65536

rabbitmq_stomp（STOMP 协议） ​

功能：支持 STOMP 协议

启用命令：

bash
rabbitmq-plugins enable rabbitmq_stomp

默认端口：61613（非加密），61614（TLS）

配置示例：

ini
# /etc/rabbitmq/rabbitmq.conf
stomp.listeners.tcp.1 = 61613
stomp.listeners.ssl.1 = 61614
stomp.default_user = guest
stomp.default_pass = guest

rabbitmq_web_stomp（WebSocket STOMP） ​

功能：通过 WebSocket 提供 STOMP 协议支持

启用命令：

bash
rabbitmq-plugins enable rabbitmq_web_stomp

默认端口：15674

访问地址：ws://localhost:15674/ws

rabbitmq_federation（联邦插件） ​

功能：实现跨集群的消息传递

启用命令：

bash
rabbitmq-plugins enable rabbitmq_federation
rabbitmq-plugins enable rabbitmq_federation_management

配置示例：

bash
# 设置上游连接
rabbitmqctl set_parameter federation-upstream my-upstream \
    '{"uri":"amqp://user:password@remote-host:5672/%2f"}'

# 设置策略
rabbitmqctl set_policy --apply-to exchanges federation-policy \
    "^federated\." \
    '{"federation-upstream":"my-upstream"}'

rabbitmq_shovel（数据迁移） ​

功能：在队列之间或集群之间迁移消息

启用命令：

bash
rabbitmq-plugins enable rabbitmq_shovel
rabbitmq-plugins enable rabbitmq_shovel_management

配置示例：

bash
rabbitmqctl set_parameter shovel my-shovel \
    '{
        "src-protocol": "amqp091",
        "src-uri": "amqp://localhost",
        "src-queue": "source-queue",
        "dest-protocol": "amqp091",
        "dest-uri": "amqp://remote-host",
        "dest-queue": "dest-queue"
    }'

rabbitmq_delayed_message_exchange（延迟消息） ​

功能：支持延迟消息投递

启用命令：

bash
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

使用示例：

php
<?php
$exchange = new AMQPExchange($channel);
$exchange->setName('delayed.exchange');
$exchange->setType('x-delayed-message');
$exchange->setArgument('x-delayed-type', 'direct');
$exchange->declareExchange();

$message = 'Delayed message';
$exchange->publish($message, 'routing.key', AMQP_NOPARAM, [
    'headers' => [
        'x-delay' => 60000  // 延迟 60 秒
    ]
]);

rabbitmq_tracing（消息追踪） ​

功能：记录消息的发送和接收日志

启用命令：

bash
rabbitmq-plugins enable rabbitmq_tracing

使用方法：

  1. 在管理界面 Admin → Tracing 中创建追踪
  2. 选择要追踪的虚拟主机
  3. 设置追踪格式（text/json）
  4. 查看追踪日志

rabbitmq_auth_backend_ldap（LDAP 认证） ​

功能：使用 LDAP 服务器进行用户认证和授权

启用命令：

bash
rabbitmq-plugins enable rabbitmq_auth_backend_ldap

配置示例：

ini
# /etc/rabbitmq/rabbitmq.conf
auth_backends.1 = ldap
auth_ldap.servers.1 = ldap.example.com
auth_ldap.port = 389
auth_ldap.user_dn_pattern = cn=${username},ou=users,dc=example,dc=com
auth_ldap.dn_lookup_bind.user_dn = cn=admin,dc=example,dc=com
auth_ldap.dn_lookup_bind.password = admin_password

rabbitmq_consistent_hash_exchange（一致性哈希） ​

功能：基于一致性哈希算法的路由

启用命令：

bash
rabbitmq-plugins enable rabbitmq_consistent_hash_exchange

使用场景：消息负载均衡到多个队列

使用示例：

php
<?php
$exchange = new AMQPExchange($channel);
$exchange->setName('hash.exchange');
$exchange->setType('x-consistent-hash');
$exchange->declareExchange();

$queue1 = new AMQPQueue($channel);
$queue1->setName('queue.1');
$queue1->declareQueue();
$queue1->bind('hash.exchange', '1');  // 权重 1

$queue2 = new AMQPQueue($channel);
$queue2->setName('queue.2');
$queue2->declareQueue();
$queue2->bind('hash.exchange', '3');  // 权重 3

$exchange->publish('message', 'user_123');

PHP 代码示例 ​

插件管理类 ​

php
<?php

class RabbitMQPluginManager
{
    private $host;
    private $sshConnection;

    public function __construct($host, $username = 'root', $password = null, $keyFile = null)
    {
        $this->host = $host;
        
        $this->sshConnection = ssh2_connect($host, 22);
        
        if ($password) {
            ssh2_auth_password($this->sshConnection, $username, $password);
        } elseif ($keyFile) {
            ssh2_auth_pubkey_file($this->sshConnection, $username, $keyFile . '.pub', $keyFile);
        }
    }

    private function execute($command)
    {
        $stream = ssh2_exec($this->sshConnection, $command);
        stream_set_blocking($stream, true);
        $output = stream_get_contents($stream);
        $errorStream = ssh2_fetch_stream($stream, SSH2_STREAM_STDERR);
        $error = stream_get_contents($errorStream);
        fclose($stream);
        
        return [
            'output' => trim($output),
            'error' => trim($error),
            'success' => empty($error)
        ];
    }

    public function listPlugins($enabledOnly = false)
    {
        $flag = $enabledOnly ? '-e' : '';
        $result = $this->execute("rabbitmq-plugins list {$flag}");
        
        $plugins = [];
        $lines = explode("\n", $result['output']);
        
        foreach ($lines as $line) {
            if (preg_match('/\[(E|e| )(\*| )\]\s+(\S+)\s+(\S+)/', $line, $matches)) {
                $plugins[] = [
                    'name' => $matches[3],
                    'version' => $matches[4],
                    'explicitly_enabled' => $matches[1] === 'E',
                    'implicitly_enabled' => $matches[1] === 'e',
                    'running' => $matches[2] === '*',
                    'enabled' => $matches[1] !== ' '
                ];
            }
        }
        
        return $plugins;
    }

    public function enablePlugin($pluginName)
    {
        return $this->execute("rabbitmq-plugins enable {$pluginName}");
    }

    public function disablePlugin($pluginName)
    {
        return $this->execute("rabbitmq-plugins disable {$pluginName}");
    }

    public function enableMultiplePlugins(array $plugins)
    {
        $pluginList = implode(' ', $plugins);
        return $this->execute("rabbitmq-plugins enable {$pluginList}");
    }

    public function disableMultiplePlugins(array $plugins)
    {
        $pluginList = implode(' ', $plugins);
        return $this->execute("rabbitmq-plugins disable {$pluginList}");
    }

    public function isPluginEnabled($pluginName)
    {
        $result = $this->execute("rabbitmq-plugins is_enabled {$pluginName}");
        return strpos($result['output'], 'is enabled') !== false;
    }

    public function enableManagement()
    {
        return $this->enablePlugin('rabbitmq_management');
    }

    public function enablePrometheus()
    {
        return $this->enablePlugin('rabbitmq_prometheus');
    }

    public function enableMQTT()
    {
        return $this->enablePlugin('rabbitmq_mqtt');
    }

    public function enableSTOMP()
    {
        return $this->enablePlugin('rabbitmq_stomp');
    }

    public function enableWebSTOMP()
    {
        return $this->enablePlugin('rabbitmq_web_stomp');
    }

    public function enableFederation()
    {
        return $this->enableMultiplePlugins([
            'rabbitmq_federation',
            'rabbitmq_federation_management'
        ]);
    }

    public function enableShovel()
    {
        return $this->enableMultiplePlugins([
            'rabbitmq_shovel',
            'rabbitmq_shovel_management'
        ]);
    }

    public function enableDelayedMessage()
    {
        return $this->enablePlugin('rabbitmq_delayed_message_exchange');
    }

    public function enableTracing()
    {
        return $this->enablePlugin('rabbitmq_tracing');
    }

    public function enableLDAP()
    {
        return $this->enablePlugin('rabbitmq_auth_backend_ldap');
    }

    public function enableConsistentHash()
    {
        return $this->enablePlugin('rabbitmq_consistent_hash_exchange');
    }

    public function getPluginStatus()
    {
        $plugins = $this->listPlugins();
        
        $status = [
            'total' => count($plugins),
            'enabled' => 0,
            'running' => 0,
            'disabled' => 0
        ];
        
        foreach ($plugins as $plugin) {
            if ($plugin['enabled']) {
                $status['enabled']++;
            } else {
                $status['disabled']++;
            }
            
            if ($plugin['running']) {
                $status['running']++;
            }
        }
        
        return $status;
    }
}

$manager = new RabbitMQPluginManager('192.168.1.100', 'admin', 'password');

$plugins = $manager->listPlugins();
echo "已安装插件总数: " . count($plugins) . "\n";

$enabledPlugins = $manager->listPlugins(true);
echo "已启用插件数: " . count($enabledPlugins) . "\n";

$result = $manager->enableManagement();
echo "启用管理插件: " . ($result['success'] ? '成功' : '失败') . "\n";

$result = $manager->enablePrometheus();
echo "启用 Prometheus 插件: " . ($result['success'] ? '成功' : '失败') . "\n";

$status = $manager->getPluginStatus();
print_r($status);

自动化插件配置脚本 ​

php
<?php

class RabbitMQPluginSetup
{
    private $pluginManager;
    private $config;

    public function __construct(RabbitMQPluginManager $manager, array $config)
    {
        $this->pluginManager = $manager;
        $this->config = $config;
    }

    public function setup()
    {
        $results = [];
        
        foreach ($this->config['plugins'] as $plugin => $settings) {
            if ($settings['enabled']) {
                $result = $this->pluginManager->enablePlugin($plugin);
                $results[$plugin] = [
                    'action' => 'enable',
                    'success' => $result['success'],
                    'message' => $result['success'] ? '启用成功' : $result['error']
                ];
            }
        }
        
        return $results;
    }

    public function validate()
    {
        $errors = [];
        
        foreach ($this->config['plugins'] as $plugin => $settings) {
            if ($settings['required'] && !$this->pluginManager->isPluginEnabled($plugin)) {
                $errors[] = "必需插件 {$plugin} 未启用";
            }
        }
        
        return $errors;
    }

    public function generateReport()
    {
        $plugins = $this->pluginManager->listPlugins();
        $status = $this->pluginManager->getPluginStatus();
        
        $report = "=== RabbitMQ 插件状态报告 ===\n";
        $report .= "时间: " . date('Y-m-d H:i:s') . "\n\n";
        $report .= "统计:\n";
        $report .= "  总插件数: {$status['total']}\n";
        $report .= "  已启用: {$status['enabled']}\n";
        $report .= "  运行中: {$status['running']}\n";
        $report .= "  未启用: {$status['disabled']}\n\n";
        
        $report .= "已启用插件:\n";
        foreach ($plugins as $plugin) {
            if ($plugin['enabled']) {
                $statusIcon = $plugin['running'] ? '✓' : '○';
                $type = $plugin['explicitly_enabled'] ? '[显式]' : '[隐式]';
                $report .= "  {$statusIcon} {$plugin['name']} {$type} v{$plugin['version']}\n";
            }
        }
        
        return $report;
    }
}

$config = [
    'plugins' => [
        'rabbitmq_management' => [
            'enabled' => true,
            'required' => true
        ],
        'rabbitmq_prometheus' => [
            'enabled' => true,
            'required' => false
        ],
        'rabbitmq_mqtt' => [
            'enabled' => true,
            'required' => false
        ],
        'rabbitmq_stomp' => [
            'enabled' => false,
            'required' => false
        ],
        'rabbitmq_federation' => [
            'enabled' => false,
            'required' => false
        ],
        'rabbitmq_delayed_message_exchange' => [
            'enabled' => true,
            'required' => false
        ]
    ]
];

$manager = new RabbitMQPluginManager('192.168.1.100', 'admin', 'password');
$setup = new RabbitMQPluginSetup($manager, $config);

$results = $setup->setup();
print_r($results);

$errors = $setup->validate();
if (!empty($errors)) {
    echo "验证错误:\n";
    foreach ($errors as $error) {
        echo "  - {$error}\n";
    }
}

echo $setup->generateReport();

实际应用场景 ​

场景一：生产环境标准配置 ​

bash
#!/bin/bash

# 生产环境 RabbitMQ 插件配置

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

# 启用 Prometheus 监控
rabbitmq-plugins enable rabbitmq_prometheus

# 启用消息追踪（可选，用于问题排查）
rabbitmq-plugins enable rabbitmq_tracing

# 验证插件状态
rabbitmq-plugins list -e

场景二：IoT 设备消息服务 ​

bash
#!/bin/bash

# IoT 场景插件配置

# 启用 MQTT 协议
rabbitmq-plugins enable rabbitmq_mqtt

# 启用 WebSocket MQTT
rabbitmq-plugins enable rabbitmq_web_mqtt

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

# 配置 MQTT
cat > /etc/rabbitmq/rabbitmq.conf << EOF
mqtt.listeners.tcp.default = 1883
mqtt.listeners.ssl.default = 8883
mqtt.allow_anonymous = false
mqtt.vhost = /iot
EOF

# 重启服务
systemctl restart rabbitmq-server

场景三：跨数据中心消息同步 ​

bash
#!/bin/bash

# 跨数据中心配置

# 启用联邦插件
rabbitmq-plugins enable rabbitmq_federation
rabbitmq-plugins enable rabbitmq_federation_management

# 启用 Shovel 插件
rabbitmq-plugins enable rabbitmq_shovel
rabbitmq-plugins enable rabbitmq_shovel_management

# 配置联邦上游
rabbitmqctl set_parameter federation-upstream dc2-upstream \
    '{"uri":"amqp://user:password@dc2.example.com:5672/%2f"}'

# 设置联邦策略
rabbitmqctl set_policy --apply-to exchanges federation-policy \
    "^federated\." \
    '{"federation-upstream":"dc2-upstream"}'

场景四：延迟任务处理 ​

bash
#!/bin/bash

# 延迟消息配置

# 启用延迟消息插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

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

# 创建延迟交换器（通过 API）
curl -u admin:password -X PUT \
    http://localhost:15672/api/exchanges/%2f/delayed.exchange \
    -H 'content-type: application/json' \
    -d '{"type":"x-delayed-message","arguments":{"x-delayed-type":"direct"}}'

常见问题与解决方案 ​

问题一：插件启用失败 ​

错误信息：

Error: The following plugins could not be found: rabbitmq_xxx

原因：插件不存在或未安装

解决方案：

bash
# 检查可用插件
rabbitmq-plugins list | grep xxx

# 安装社区插件
# 下载 .ez 文件到 /usr/lib/rabbitmq/plugins/
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.12.0/rabbitmq_delayed_message_exchange-3.12.0.ez
mv rabbitmq_delayed_message_exchange-3.12.0.ez /usr/lib/rabbitmq/plugins/

# 启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

问题二：插件依赖问题 ​

错误信息：

Error: Dependencies failed

解决方案：

bash
# 启用时会自动处理依赖
rabbitmq-plugins enable rabbitmq_management

# 查看依赖关系
rabbitmq-plugins list -v | grep rabbitmq_management

问题三：插件禁用后服务异常 ​

原因：禁用了核心依赖插件

解决方案：

bash
# 重新启用必要插件
rabbitmq-plugins enable rabbitmq_management

# 检查服务状态
systemctl status rabbitmq-server

问题四：插件配置不生效 ​

原因：配置文件错误或未重启服务

解决方案：

bash
# 检查配置文件语法
rabbitmqctl eval 'application:get_env(rabbit, mqtt).'

# 重启服务
systemctl restart rabbitmq-server

# 检查插件状态
rabbitmq-plugins list -e

问题五：离线模式启用插件 ​

场景：节点未启动时需要启用插件

解决方案：

bash
# 使用离线模式
rabbitmq-plugins enable --offline rabbitmq_management

最佳实践建议 ​

1. 插件选择原则 ​

  • 只启用必需的插件
  • 避免启用实验性插件
  • 定期审查插件使用情况
  • 关注插件版本兼容性

2. 生产环境建议 ​

  • 启用管理界面便于监控
  • 启用 Prometheus 插件集成监控
  • 考虑启用消息追踪用于问题排查
  • 根据业务需求选择协议插件

3. 安全配置 ​

  • 禁用不必要的协议插件
  • 配置认证后端插件
  • 使用 TLS 加密连接
  • 定期更新插件版本

4. 性能优化 ​

  • 监控插件资源使用
  • 合理配置插件参数
  • 避免过多插件影响性能
  • 使用延迟消息时注意队列积压

5. 运维管理 ​

  • 记录插件配置变更
  • 使用配置文件管理插件
  • 建立插件启用/禁用流程
  • 准备回滚方案

相关链接 ​