容量规划与评估

本文介绍RabbitMQ容量规划方法，通过业务指标测算集群规模与资源配置。

定义

容量规划是根据业务消息吞吐量、延迟要求、持久化需求等指标，通过性能基准测试与资源测算，确定RabbitMQ集群节点数量、硬件配置、网络带宽与存储容量的工程方法。

原理

容量测算维度

业务指标输入:
├─ TPS: 每秒消息发布/消费量
├─ 消息大小: 平均/最大消息体积
├─ 持久化比例: 需要持久化的消息占比
├─ 延迟要求: P99延迟上限
└─ 保留时间: 消息最长留存时间

资源输出:
├─ CPU核数: 网络IO+序列化+路由计算
├─ 内存容量: 活跃消息+连接+通道缓存
├─ 磁盘容量: 持久化消息+索引+mnesia
└─ 网络带宽: 发布+消费+集群同步流量

测算公式

内存需求:
  memory = (活跃消息数 × 平均大小) + (连接数 × 64KB) + (通道数 × 16KB) + 基础开销(256MB)

磁盘需求:
  disk = (日持久化消息量 × 平均大小 × 保留天数) / 压缩率(约0.6) + mnesia(2GB)

网络带宽:
  bandwidth = (TPS_publish × 消息大小 × 8) + (TPS_consume × 消息大小 × 8) + 集群同步(20%)

CPU需求:
  CPU_cores = (TPS_total / 单核处理能力) × 1.5(冗余系数)
  单核处理能力: 约1万TPS(1KB消息, 持久化)

集群规模评估

示例

Maven依赖

<dependencies>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.20.0</version>
    </dependency>
</dependencies>

容量计算器

import java.text.DecimalFormat;

public class CapacityPlanner {
    private static final DecimalFormat df = new DecimalFormat("#.##");

    public static void main(String[] args) {
        // 业务指标输入
        long tpsPublish = 5000;     // 每秒发布
        long tpsConsume = 5000;     // 每秒消费
        long msgSizeAvg = 1024;     // 平均1KB
        double persistRatio = 0.8;  // 80%持久化
        int retentionDays = 3;      // 保留3天
        int connections = 200;      // 连接数
        int channels = 1000;        // 通道数

        // 内存计算
        long activeMessages = tpsPublish * 60; // 假设60秒消费延迟
        long memoryMB = (activeMessages * msgSizeAvg) / (1024 * 1024)
            + (connections * 64) / (1024 * 1024)
            + (channels * 16) / (1024 * 1024)
            + 256;
        System.out.println("Memory required: " + memoryMB + " MB");

        // 磁盘计算
        long dailyPersisted = (long)(tpsPublish * 86400 * persistRatio);
        long diskGB = (dailyPersisted * msgSizeAvg * retentionDays) / (1024 * 1024 * 1024);
        diskGB = (long)(diskGB / 0.6); // 压缩率
        diskGB += 2; // mnesia
        System.out.println("Disk required: " + diskGB + " GB");

        // 网络带宽计算
        double bandwidthMbps = ((tpsPublish + tpsConsume) * msgSizeAvg * 8) / (1024 * 1024.0);
        bandwidthMbps *= 1.2; // 集群同步20%
        System.out.println("Network bandwidth: " + df.format(bandwidthMbps) + " Mbps");

        // CPU计算
        double cpuCores = (double)(tpsPublish + tpsConsume) / 10000 * 1.5;
        System.out.println("CPU cores: " + df.format(cpuCores));

        // 推荐配置
        recommendCluster(cpuCores, memoryMB, diskGB);
    }

    private static void recommendCluster(double cpuCores, long memoryMB, long diskGB) {
        System.out.println("\n=== Recommended Cluster ===");
        if (cpuCores <= 4 && memoryMB <= 8192) {
            System.out.println("Nodes: 2");
            System.out.println("Spec: 4C8G, 100GB SSD per node");
        } else if (cpuCores <= 8 && memoryMB <= 16384) {
            System.out.println("Nodes: 3");
            System.out.println("Spec: 8C16G, 500GB SSD per node");
        } else {
            int nodes = (int)Math.ceil(cpuCores / 8) + 1;
            System.out.println("Nodes: " + nodes);
            System.out.println("Spec: 16C32G, 1TB NVMe per node");
        }
    }
}

性能基准测试

import com.rabbitmq.client.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class BenchmarkTest {
    private static final int THREADS = 10;
    private static final int MESSAGES = 50000;
    private static final int MSG_SIZE = 1024;

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setRequestedHeartbeat(30);

        AtomicLong totalLatency = new AtomicLong(0);
        CountDownLatch latch = new CountDownLatch(THREADS);
        ExecutorService executor = Executors.newFixedThreadPool(THREADS);

        long start = System.currentTimeMillis();

        for (int t = 0; t < THREADS; t++) {
            executor.submit(() -> {
                try (Connection conn = factory.newConnection();
                     Channel ch = conn.createChannel()) {
                    ch.confirmSelect(); // 开启confirm模式
                    ch.queueDeclare("benchmark_queue", true, false, false, null);

                    byte[] payload = new byte[MSG_SIZE];
                    for (int i = 0; i < MESSAGES; i++) {
                        long msgStart = System.nanoTime();
                        ch.basicPublish("", "benchmark_queue", null, payload);
                        ch.waitForConfirms(); // 等待confirm
                        long msgEnd = System.nanoTime();
                        totalLatency.addAndGet(msgEnd - msgStart);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    latch.countDown();
                }
            });
        }

        latch.await();
        long duration = System.currentTimeMillis() - start;
        long totalMsgs = THREADS * MESSAGES;

        System.out.println("Total messages: " + totalMsgs);
        System.out.println("Duration: " + duration + " ms");
        System.out.println("TPS: " + (totalMsgs * 1000L / duration));
        System.out.println("Avg latency: " + (totalLatency.get() / totalMsgs / 1000_000) + " ms");

        executor.shutdown();
    }
}

容量监控脚本

#!/bin/bash
# capacity_monitor.sh

HOST="localhost"
PORT=15672
USER="admin"
PASS="admin"

echo "=== RabbitMQ Capacity Report ==="
echo ""

# 内存使用
echo "Memory Usage:"
curl -s -u $USER:$PASS http://$HOST:$PORT/api/nodes | \
    python3 -c "import sys,json; data=json.load(sys.stdin);
for n in data: print(f\"  {n['name']}: {n['mem_used']/1024/1024:.0f}MB / {n['mem_limit']/1024/1024:.0f}MB\")"

# 磁盘使用
echo ""
echo "Disk Usage:"
curl -s -u $USER:$PASS http://$HOST:$PORT/api/nodes | \
    python3 -c "import sys,json; data=json.load(sys.stdin);
for n in data: print(f\"  {n['name']}: {n['disk_free']/1024/1024:.0f}MB free\")"

# 队列消息量
echo ""
echo "Queue Message Count:"
curl -s -u $USER:$PASS http://$HOST:$PORT/api/queues | \
    python3 -c "import sys,json; data=json.load(sys.stdin);
for q in data: print(f\"  {q['name']}: {q.get('messages',0)} messages\")"

# 连接数
echo ""
echo "Connections:"
curl -s -u $USER:$PASS http://$HOST:$PORT/api/overview | \
    python3 -c "import sys,json; data=json.load(sys.stdin);
print(f\"  Total: {data.get('object_totals',{}).get('connections',0)}\")"

注意事项

容量规划必须包含20%-30%冗余，否则峰值流量会导致性能骤降。

基准测试应在生产相似环境执行，否则测算结果偏差较大。

磁盘容量需考虑消息压缩率，否则实际存储量会高于预期。

内存计算中的"活跃消息"指未被消费的消息，而非队列总消息量。

集群节点数推荐奇数（3或5），否则脑裂风险增加。

容量规划需定期复核，业务增长后需及时扩容。

要点总结

• 容量规划通过业务指标输入，测算CPU、内存、磁盘、网络需求
• 测算公式基于活跃消息量、持久化比例、连接通道数
• 集群规模根据TPS与资源需求选择2-5+节点
• 基准测试使用confirm模式验证真实TPS与延迟
• 容量规划必须包含冗余，定期复核并及时扩容