性能优化手册

# 前言

本文档介绍金蝶Apusic分布式消息队列for RabbitMQ（Apusic Distributed Message Queue，简称：ADMQ for RabbitMQ）的性能优化最佳实践，帮助用户充分利用系统性能。

# 适用对象

本文档适用于IT信息化业务负责人、研发经理、软件项目经理、软件架构师、运维工程师。

# 相关文档

了解更多ADMQ for RabbitMQ产品相关的信息，请参阅以下ADMQ for RabbitMQ产品手册文档集：

序号 手册文档 说明
1 金蝶Apusic分布式消息队列for RabbitMQ 快速使用手册 简单介绍了如何快速上手使用ADMQ for RabbitMQ 。
2 金蝶Apusic分布式消息队列for RabbitMQ 安装手册 详细介绍如何在各操作系统上安装ADMQ for RabbitMQ，以及ADMQ for RabbitMQ服务启停等操作。
3 金蝶Apusic分布式消息队列for RabbitMQ 消息引擎用户手册 详细介绍 ADMQ for RabbitMQ 消息引擎相关功能的使用、配置、管理及配套工具的使用方法。
4 金蝶Apusic分布式消息队列for RabbitMQ 管控台用户手册 详细介绍ADMQ for RabbitMQ管控台相关功能的使用和操作说明。
5 金蝶Apusic分布式消息队列for RabbitMQ 开发手册 详细介绍基于各开发语言进行ADMQ for RabbitMQ客户端应用开发的说明。
6 金蝶Apusic分布式消息队列for RabbitMQ 迁移手册 详细介绍从RabbitMQ迁移到ADMQ for RabbitMQ的说明。
7 金蝶Apusic分布式消息队列for RabbitMQ 运维手册 详细介绍ADMQ for RabbitMQ的监控、运维、安全加固等运维说明。
8 金蝶Apusic分布式消息队列for RabbitMQ 性能优化手册 详细介绍ADMQ for RabbitMQ性能调优的说明。

# 技术支持

ADMQ for RabbitMQ产品提供全面的技术支持服务，您可以通过以下方式获得技术支持：

  • 网址：www.apusic.com
  • 电话：400-855-5800
  • 邮箱：support@apusic.com
  • 金蝶云社区：https://vip.kingdee.com/?productId=73&productLineId=14&lang=zh-CN

您在取得技术支持时，请提供如下信息：

  1. 您的姓名

  2. 公司信息与联系方式

  3. 操作系统及其版本

  4. 产品版本号

  5. 出现异常及错误的日志、截图等详细信息

# 系统层面优化

# 操作系统优化

# 文件描述符

# 查看当前限制
ulimit -n

# 修改限制（/etc/security/limits.conf）
* soft nofile 65535
* hard nofile 65535
1
2
3
4
5
6

# 内核参数

# /etc/sysctl.conf

# 增加端口范围
net.ipv4.ip_local_port_range = 1024 65535

# TCP 连接优化
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 30
net.core.somaxconn = 65535

# 内存优化
vm.swappiness = 10
vm.dirty_ratio = 40
vm.dirty_background_ratio = 10

# 应用配置
sysctl -p
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 磁盘优化

优化项 建议
文件系统 使用 XFS 文件系统
挂载选项 noatime,nodiratime
磁盘调度 使用 noop 或 deadline
RAID 配置 使用 RAID 10 或 RAID 0
SSD 优化 启用 TRIM，预留 over-provisioning
# 查看磁盘调度算法
cat /sys/block/sda/queue/scheduler

# 设置为 deadline
echo deadline > /sys/block/sda/queue/scheduler
1
2
3
4
5
6

# 网络优化

# 网络缓冲区
net.core.rmem_max = 134217728
net.core.wmem_max = 134217728
net.ipv4.tcp_rmem = 4096 87380 134217728
net.ipv4.tcp_wmem = 4096 65536 134217728

# TCP 优化
net.ipv4.tcp_window_scaling = 1
net.ipv4.tcp_timestamps = 1
net.ipv4.tcp_sack = 1
net.ipv4.tcp_no_metrics_save = 1
1
2
3
4
5
6
7
8
9
10
11

# RabbitMQ 配置优化

# 内存管理

参数 默认值 推荐值 说明
vm_memory_high_watermark.relative 0.4 0.6-0.7 内存使用上限比例
vm_memory_high_watermark_paging_ratio 0.5 0.5 开始分页的内存比例
vm_memory_high_watermark.absolute - 4G 绝对内存上限
# rabbitmq.conf
vm_memory_high_watermark.relative = 0.6
vm_memory_high_watermark_paging_ratio = 0.5
1
2
3

# 磁盘管理

参数 默认值 推荐值 说明
disk_free_limit.relative 2.0 1.5 磁盘空间下限比例
disk_free_limit.absolute 50MB 1GB 绝对磁盘空间下限
# rabbitmq.conf
disk_free_limit.relative = 1.5
disk_free_limit.absolute = 1GB
1
2
3

# 连接与通道优化

参数 默认值 推荐值 说明
channel_max 2047 根据需求调整 每个连接的最大通道数
connection_max infinity 根据需求调整 最大连接数
heartbeat 60 30-60 心跳间隔（秒）
handshake_timeout 10000 10000 握手超时（毫秒）
# rabbitmq.conf
channel_max = 2047
heartbeat = 30
handshake_timeout = 10000
1
2
3
4

# 队列优化

# 队列类型选择

队列类型 吞吐量 可靠性 适用场景
Classic Queue 中等 中（需配置镜像） 一般业务场景
Quorum Queue 中等 高 高可用场景
Stream Queue 极高 高 高吞吐、日志场景

# 队列参数优化

Map<String, Object> args = new HashMap<>();

# 队列长度限制，防止无限堆积
args.put("x-max-length", 100000);
args.put("x-max-length-bytes", 1073741824);  # 1GB

# 消息 TTL
args.put("x-message-ttl", 3600000);  # 1小时

# 队列 TTL
args.put("x-expires", 7200000);  # 2小时无消费者后删除

# 惰性队列（减少内存使用）
args.put("x-queue-mode", "lazy");

channel.queueDeclare("optimized.queue", true, false, false, args);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 消息持久化优化

# 何时使用持久化

场景 建议
关键业务消息 必须持久化（delivery_mode=2）
日志消息 可非持久化
高吞吐场景 权衡可靠性和性能

# 批量持久化

使用 Confirm 模式 + 批量确认提高持久化性能：

channel.confirmSelect();
channel.setConfirmCallback(...);

# 批量发送
for (int i = 0; i < 1000; i++) {
    channel.basicPublish(EXCHANGE, ROUTING_KEY, 
        MessageProperties.PERSISTENT_TEXT_PLAIN, 
        message.getBytes());
}

# 等待全部确认
channel.waitForConfirmsOrDie(5000);
1
2
3
4
5
6
7
8
9
10
11
12

# 客户端优化

# 连接管理

# 连接池

# 使用连接池管理连接
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setConnectionTimeout(30000);
factory.setRequestedHeartbeat(30);
factory.setAutomaticRecoveryEnabled(true);

# 单连接多通道（推荐）
Connection connection = factory.newConnection();
for (int i = 0; i < 10; i++) {
    Channel channel = connection.createChannel();
    # 使用 channel 处理消息
}
1
2
3
4
5
6
7
8
9
10
11
12
13

# 自动恢复

factory.setAutomaticRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(5000);
factory.setTopologyRecoveryEnabled(true);
factory.setChannelRpcTimeout(10000);
1
2
3
4

# 生产者优化

# 批量发送

# 批量发送消息
List<Message> messages = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
    messages.add(new Message(...));
}

# 使用事务或 Confirm 模式保证可靠性
channel.confirmSelect();
for (Message msg : messages) {
    channel.basicPublish(EXCHANGE, ROUTING_KEY, null, msg.toBytes());
}
channel.waitForConfirmsOrDie();
1
2
3
4
5
6
7
8
9
10
11
12

# 异步发送

# 使用线程池异步发送
ExecutorService executor = Executors.newFixedThreadPool(10);
for (Message msg : messages) {
    executor.submit(() -> {
        channel.basicPublish(EXCHANGE, ROUTING_KEY, null, msg.toBytes());
    });
}
1
2
3
4
5
6
7

# 消费者优化

# QoS 预取

# 设置 QoS，控制未确认消息数量
channel.basicQos(100);  # 每个消费者最多 100 条未确认消息

# 全局 QoS
channel.basicQos(0, 100, true);  # 整个通道最多 100 条
1
2
3
4
5

# 批量确认

# 消费消息并批量确认
List<Long> deliveryTags = new ArrayList<>();
for (int i = 0; i < 100; i++) {
    GetResponse response = channel.basicGet(QUEUE, false);
    if (response != null) {
        deliveryTags.add(response.getEnvelope().getDeliveryTag());
        processMessage(response);
    }
}

# 批量确认
if (!deliveryTags.isEmpty()) {
    channel.basicAck(deliveryTags.get(deliveryTags.size() - 1), true);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 多消费者

# 创建多个消费者并行处理
for (int i = 0; i < 10; i++) {
    Channel channel = connection.createChannel();
    channel.basicQos(50);
    
    DeliverCallback callback = (tag, delivery) -> {
        processMessage(delivery);
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    };
    
    channel.basicConsume(QUEUE, false, callback, tag -> {});
}
1
2
3
4
5
6
7
8
9
10
11
12

# 集群优化

# 节点部署策略

部署策略 说明 适用场景
同机房部署 所有节点在同一机房 低延迟、高吞吐
跨机房部署 节点分布在不同机房 高可用、容灾
异构部署 不同配置节点混合 成本优化

# 镜像队列优化

# 优化镜像同步策略
bin/admq rabbitmq admin ctl set_policy ha-optimized "^important-.*" '
{
    "ha-mode": "exactly",
    "ha-params": 2,
    "ha-sync-mode": "automatic",
    "ha-promote-on-shutdown": "when-synced"
}'
1
2
3
4
5
6
7
8

# Quorum Queue 优化

# 调整 Raft 参数
bin/admq rabbitmq admin ctl eval 'application:set_env(rabbit, quorum_commands_soft_limit, 256).'
bin/admq rabbitmq admin ctl eval 'application:set_env(rabbit, quorum_commands_hard_limit, 512).'
1
2
3

# Stream Queue 优化

Map<String, Object> args = new HashMap<>();
# 段文件大小
args.put("x-stream-max-segment-size-bytes", 104857600);  # 100MB

# 保留策略
args.put("x-stream-max-retention-size-bytes", 10737418240L);  # 10GB
args.put("x-stream-max-retention-time", 604800000);  # 7天

channel.queueDeclare("stream.queue", true, false, false, args);
1
2
3
4
5
6
7
8
9

# 监控与调优

# 性能指标

指标 健康范围 优化建议
CPU 使用率 < 70% 扩容或优化业务逻辑
内存使用率 < vm_memory_high_watermark 增加内存或优化队列配置
磁盘使用率 < 80% 扩容或清理数据
消息入速率 根据业务需求 优化生产者或扩容
消息出速率 接近入速率 优化消费者或增加消费者
未确认消息 < QoS 预取数 优化消费者处理速度
连接数 < connection_max 优化连接使用或使用连接池

# 性能测试

# 使用 perf-test 工具测试
rabbitmq-perf-test -x 10 -y 10 -u "test-queue" -a --auto-delete false

# 参数说明
# -x: 生产者数量
# -y: 消费者数量
# -u: 队列名称
# -a: 自动确认
# --auto-delete: 是否自动删除队列
1
2
3
4
5
6
7
8
9

# 瓶颈分析

  1. CPU 瓶颈：消息处理逻辑复杂、加密解密、大量连接
  2. 内存瓶颈：队列堆积、大量连接、未启用惰性队列
  3. 磁盘瓶颈：大量持久化消息、磁盘 I/O 不足
  4. 网络瓶颈：跨机房部署、大量数据传输

# 常见问题优化

# 消息堆积优化

  1. 增加消费者数量
  2. 优化消费者处理逻辑
  3. 设置队列长度限制
  4. 使用死信队列处理过期消息

# 内存不足优化

  1. 增加物理内存
  2. 调整 vm_memory_high_watermark
  3. 启用惰性队列
  4. 清理不必要的队列和消息
  5. 减少连接数

# 高延迟优化

  1. 使用自动确认（非关键消息）
  2. 减少持久化频率
  3. 优化网络配置
  4. 使用更近的节点
  5. 减少消息大小

# 性能优化检查清单

# 系统层面

  • [ ] 文件描述符限制已调整
  • [ ] 内核参数已优化
  • [ ] 磁盘使用 XFS 文件系统
  • [ ] 网络参数已优化
  • [ ] 系统资源充足（CPU、内存、磁盘）

# RabbitMQ 配置

  • [ ] 内存水位线合理设置
  • [ ] 磁盘空间限制合理设置
  • [ ] 心跳间隔适当
  • [ ] 连接和通道数限制合理
  • [ ] 队列 Leader 分配策略适当

# 客户端优化

  • [ ] 使用连接池
  • [ ] 使用单连接多通道
  • [ ] 启用自动恢复
  • [ ] 合理设置 QoS
  • [ ] 使用批量确认
  • [ ] 选择合适的确认模式

# 集群优化

  • [ ] 节点选择合适的队列类型
  • [ ] 镜像队列策略合理
  • [ ] Quorum Queue 参数优化
  • [ ] Stream Queue 保留策略合理
  • [ ] 集群规模满足业务需求
编辑页面 (opens new window)
#性能优化手册

← 极简运维手册→