4. 吞吐量定义：消息吞吐量的计算公式、单位（TPS、MB/s）与影响因素
吞吐量是衡量消息队列处理能力最核心的量化指标，直接反映系统在单位时间内能承载的业务消息规模。在读写分离架构下，吞吐量需要分别从生产端（写入）和消费端（读取）两个维度进行定义与测量，才能准确评估分离后的性能增益。
4.1 吞吐量的计算公式
消息吞吐量的计算遵循“总量 / 时间窗口”的基本范式，根据统计口径不同分为以下两种常用公式：
| 指标 | 计算公式 | 说明 |
|---|---|---|
| 消息条数吞吐量（TPS） | TPS = 成功处理的消息总条数 / 测试持续时间（秒） |
适用于消息体大小相对固定的场景，如订单、支付通知 |
| 数据量吞吐量（MB/s） | MB/s = 成功处理的消息总字节数 / (测试持续时间秒 × 1024 × 1024) |
适用于消息体大小波动大的场景，如日志、图片元数据 |
实践要点： 在读写分离测试中，应同时记录 TPS 和 MB/s。若消息体平均大小超过 10 KB，MB/s 往往比 TPS 更能反映网络和磁盘的真实瓶颈。
4.2 吞吐量的单位体系
吞吐量单位的选择取决于测试目标和消息特征。下表列出读写分离场景下最常用的单位及其适用场景：
| 单位 | 全称 | 典型数值范围 | 适用场景 |
|---|---|---|---|
| TPS | Transactions Per Second | 1,000 ~ 1,000,000+ | 金融交易、实时通知、RPC 解耦 |
| MB/s | Megabytes Per Second | 10 ~ 5,000+ | 日志采集、大数据管道、视频流元数据 |
| msg/s | Messages Per Second | 与 TPS 等价 | 通用场景，与 TPS 可互换使用 |
| K msg/s | Kilo Messages Per Second | 1 ~ 1,000 | 高吞吐场景下的简化表示 |
单位换算示例： 若消息平均大小为 2 KB，测得 50,000 TPS，则 MB/s = 50,000 × 2 KB / 1024 ≈ 97.66 MB/s。读写分离后，写入节点和读取节点应分别计算此值。
4.3 吞吐量的核心影响因素
在读写分离架构下，吞吐量受多个层次的因素制约。下图展示了从生产者到消费者的完整链路及各环节的关键瓶颈：
4.4 读写分离场景下的吞吐量关键影响因素
在读写分离架构中，以下因素对吞吐量的影响尤为显著，测试时应重点监控：
- 消息体大小（Payload Size）：消息体越大，单条消息的序列化、网络传输和磁盘写入耗时越长。测试表明，当消息体从 1 KB 增长到 100 KB 时，TPS 可能下降 80% 以上，而 MB/s 可能仅下降 20%。
- 分区数（Partitions）：分区是并行度的基础。写入节点和读取节点的分区数配置直接影响并发吞吐。通常建议分区数不低于 CPU 核心数的 2 倍。
- 批量发送/消费大小（Batch Size）：批量操作能显著提升吞吐。生产端批量发送（如每次 500 条）可将 TPS 提升 3~5 倍；消费端批量拉取（如每次 1000 条）可减少网络往返次数。
- 副本同步策略（Replication Factor & Ack）：写入节点需要将消息同步到副本。若设置
acks=all，每次写入需等待所有副本确认，吞吐量通常下降 30%~50%。 - 磁盘类型与 IOPS：NVMe SSD 的随机写吞吐可达 500 MB/s 以上，而 HDD 通常低于 200 MB/s。读写分离后，读取节点可配置更高 IOPS 的磁盘以应对消费流量。
常见误区： 不要只关注 TPS 而忽略 MB/s。在日志场景中，消息体可能达到 50 KB，此时 10,000 TPS 对应 500 MB/s，远超千兆网卡（约 125 MB/s）的承载能力，网络会成为实际瓶颈。
4.5 吞吐量测试的基准条件
为确保吞吐量数据的可重复性和可比性，测试前必须明确以下基准条件：
| 条件项 | 推荐配置/说明 |
|---|---|
| 消息体大小 | 固定为 1 KB、10 KB、100 KB 三档分别测试 |
| 分区数 | 写入节点 8 分区，读取节点 8 分区（与 CPU 核心数匹配） |
| 批量大小 | 生产端 500 条/批，消费端 1000 条/批 |
| 确认级别 | 分别测试 acks=1 和 acks=all 两种模式 |
| 测试持续时间 | 预热 30 秒后，持续测试 120 秒，取稳定段平均值 |
| 并发线程数 | 从 1 开始逐步增加，直到吞吐不再线性增长（找到饱和点） |
测试脚本示例（伪代码）：
# 生产端吞吐测试
for batch_size in [100, 500, 1000]:
for ack_mode in [1, -1]: # -1 即 acks=all
start = now()
send_messages(batch_size, ack_mode, duration=120s)
elapsed = now() - start
tps = total_messages / elapsed
mbps = total_bytes / elapsed / 1024 / 1024
record(tps, mbps)
# 消费端吞吐测试
for fetch_size in [500, 1000, 2000]:
start = now()
consume_messages(fetch_size, duration=120s)
elapsed = now() - start
tps = total_messages / elapsed
mbps = total_bytes / elapsed / 1024 / 1024
record(tps, mbps)