• 分布式文件系统
• 对象存储
• 存储

本文导读：本文以 SeaweedFS 仓库中 test/mq/README.md 为骨架，系统讲解如何搭建 SeaweedMQ Broker/Agent 环境，构建基于结构化消息（schema-based RecordValue）的生产者与消费者测试程序，并通过 Makefile 自动化脚本完成基础收发、性能压测、多消费者组负载均衡、offset 回放等典型场景。读完本文，你将掌握从"启动 MQ 服务"到"跑通端到端收发"再到"深入源码理解底层原理"的完整链路。

一、测试套件总览

SeaweedFS 在仓库的test/mq目录下提供了一套专门用于验证 SeaweedMQ 功能的最小可运行测试程序，包含消息生产者（producer）与消息消费者（consumer）两个独立的可执行程序，以及一个用于构建、运行和测试自动化的 Makefile。

测试套件的核心设计意图在于：

1. 验证 Agent 中间层架构：客户端不直接连接 broker，而是统一通过 MQ agent 这个中间层完成发布与订阅；
2. 验证结构化消息协议：消息采用 schema 驱动的RecordValue格式，而非裸字节流，覆盖了 SeaweedMQ 区别于传统 MQ 的关键设计；
3. 验证消息生命周期全流程：从 topic 创建、分区配置、结构化消息发布，到消费者组订阅、offset 管理与滑动窗口并发消费，全链路可观测。

目录结构如下：

test/mq/ ├── producer/main.go # 消息生产者实现 ├── consumer/main.go # 消息消费者实现 ├── Makefile # 构建与测试自动化 └── README.md # 本文档（测试说明）

二、前置条件

在运行测试套件之前，需要准备：

1. 运行中的 SeaweedFS MQ 服务：需要同时启用 MQ broker 与 MQ agent 的 SeaweedFS 实例。注意，agent 依赖 broker 才能工作，二者必须同时在线；
2. Go 工具链：Go 1.19 或更高版本，用于编译测试程序（源码见 test/mq/producer/main.go 与 test/mq/consumer/main.go，编译依赖均来自仓库根目录的 go.mod）。

需要说明的是，weed/mq/README.md 中明确标注 SeaweedMQ 目前处于 "WIP, not ready" 状态，即仍在开发迭代中，本文所述测试流程适用于该仓库当前代码版本，读者在正式环境使用前应关注其演进。

三、快速开始：启动 MQ 服务并跑通首个测试

3.1 启动 SeaweedFS 的 MQ Broker 与 Agent

SeaweedMQ 采用了"broker 只计算不存储"的架构设计：broker 是无状态计算节点，消息数据实际落在 volume server 上，因此"无限扩展 broker"与"无限扩展存储"可以独立进行。测试环境可以通过两种方式启动服务：

方式一：单命令一键启动全部组件

# 同时启动 master、volume、filer、MQ broker 与 MQ agent weed server -mq.broker -mq.agent -filer -volume -master.peers=none

方式二：分组件逐一启动

weed master -peers=none weed volume -master=localhost:9333 weed filer -master=localhost:8888 weed mq.broker -filer=localhost:8888 weed mq.agent -brokers=localhost:17777

从仓库源码 weed/command/server.go 可以看到weed server通过-mq.broker与-mq.agent两个布尔开关决定是否拉起 MQ 组件，相关端口参数为：

对应单独命令行的默认端口为：mq.broker默认 17777、mq.agent默认 16777（见 weed/command/mq_broker.go 与 weed/command/mq_agent.go）。

3.2 构建测试程序

# 构建 producer 与 consumer 两个二进制 make build # 或分别构建 make build-producer make build-consumer

构建产物输出到test/mq/bin/目录。Makefile 使用go build -o bin/$(1) $(2)封装构建命令（见 test/mq/Makefile）。

3.3 运行基础测试

# 一键执行基础 producer/consumer 测试 make test # 或手动分两步执行 make consumer & # 终端 1：后台启动消费者 make producer # 终端 2：启动生产者

make test目标的核心执行逻辑是：后台拉起消费者（offset 设为earliest），等待 2 秒后启动生产者，生产者发完消息后再等待 5 秒让消费者处理，最后杀掉消费者进程（见 test/mq/Makefile）。

四、测试程序详解：Producer 与 Consumer

4.1 Producer（生产者）

Producer 程序负责生成结构化消息并通过 MQ agent 发布到指定 topic，源码位于 test/mq/producer/main.go。

使用方法：

./bin/producer [options]

命令行参数：

示例：

./bin/producer -agent=localhost:16777 -namespace=test -topic=my-topic -messages=1000 -interval=50ms

结构化消息的关键实现：Producer 定义了一个TestMessage结构体（含 ID、Message、Payload、Timestamp 四个字段），并通过schema.StructToSchema(messageInstance)利用反射自动从 Go 结构体生成RecordTypeschema，而非手工编写 schema：

type TestMessage struct { ID int64 `json:"id"` Message string `json:"message"` Payload []byte `json:"payload"` Timestamp int64 `json:"timestamp"` } messageInstance := TestMessage{} recordType := schema.StructToSchema(messageInstance)

从源码看，StructToSchema在 weed/mq/schema/struct_to_schema.go 中实现，内部通过reflect.TypeOf(instance)判断类型，并递归将 Go 基础类型映射为 schema 标量类型：bool→BOOL、int/int8/int16/int32→INT32、int64→INT64、float→FLOAT、double→DOUBLE、bytes→BYTES、string→STRING，复合类型则映射为 list/record。之后每条消息通过session.PublishMessageRecord(key, record)以"key + RecordValue"的形式发布，key 用于分区路由。

发布会话的建立：agent_client.NewPublishSession（见 weed/mq/client/agent_client/publish_session.go）通过 gRPC 连接 agent，先发送StartPublishSessionRequest（携带 topic、分区数、RecordType、publisher 名）获取 sessionId，再开启一条双向流PublishRecord，将消息一条条流式发出，最后CloseSession关闭流。

4.2 Consumer（消费者）

Consumer 程序通过 MQ agent 订阅 topic 中的结构化消息，源码位于 test/mq/consumer/main.go。

使用方法：

./bin/consumer [options]

命令行参数：

示例：

./bin/consumer -agent=localhost:16777 -namespace=test -topic=my-topic -group=my-group -offset=earliest

offset 类型的底层映射：Consumer 程序将命令行 offset 字符串映射到 schema 层的枚举：

订阅会话的关键实现：Consumer 构造agent_client.SubscribeOption（见 weed/mq/client/agent_client/subscribe_session.go）后调用NewSubscribeSession建立双向流，初始化请求中携带消费者组、实例 ID、topic、offset 类型、offset 时间戳、过滤条件、最大订阅分区数与滑动窗口大小。随后通过SubscribeMessageRecord循环接收消息并回调两个函数：

session.SubscribeMessageRecord( // onEachMessageFn：每条消息触发一次 func(key []byte, record *schema_pb.RecordValue) { mu.Lock(); messageCount++; currentCount := messageCount; mu.Unlock() if *showMessages { fmt.Printf("Received message: key=%s\n", string(key)) ; printRecordValue(record) } if *logProgress && currentCount%10 == 0 { rate := float64(currentCount) / time.Since(startTime).Seconds() fmt.Printf("Consumed %d messages (%.2f msg/sec)\n", currentCount, rate) } }, // onCompletionFn：订阅结束时触发 func() { fmt.Printf("Subscription completed\n"); done <- nil }, )

程序还监听SIGINT/SIGTERM信号实现优雅退出，退出前打印总消费量与平均吞吐，方便在后台测试场景下人工中断并观察统计结果。

五、Makefile 命令速查

构建类

运行类

测试类

其他

六、环境变量配置

Makefile 中的运行/测试目标均支持通过环境变量覆盖默认参数（默认值定义见 test/mq/Makefile）：

export AGENT_ADDR=localhost:16777 export TOPIC_NAMESPACE=test export TOPIC_NAME=test-topic export PARTITION_COUNT=4 export MESSAGE_COUNT=100 export CONSUMER_GROUP=test-consumer-group export CONSUMER_INSTANCE=test-consumer-1

也可以在命令行临时指定，例如make producer MESSAGE_COUNT=1000 PARTITION_COUNT=8，或将 agent 指向远端机器如make test AGENT_ADDR=10.21.152.113:16777 MESSAGE_COUNT=500。

七、典型应用场景演练

场景 1：基础收发测试

# 终端 1：启动消费者 make consumer # 终端 2：启动生产者（发送 50 条） make producer MESSAGE_COUNT=50

由于默认 offset 为latest，生产者在消费者启动之后才开始发送，因此消费者能够完整收到消息。

场景 2：性能压测

make test-performance

该目标内部以perf-testtopic、8 个分区、1000 条消息、每条 512 字节 payload、10ms 间隔执行高吞吐测试（见 test/mq/Makefile）。消费者侧会实时打印每 10 条消息的消费速率（msg/sec），可用于初步观察吞吐表现。

场景 3：多个消费者组并行消费

# 终端 1：消费者组 1 make consumer CONSUMER_GROUP=group1 # 终端 2：消费者组 2 make consumer CONSUMER_GROUP=group2 # 终端 3：生产者（发送 200 条） make producer MESSAGE_COUNT=200

不同消费者组各自拥有独立的消费进度（offset），因此每个组都能消费到全部 200 条消息；而在同一消费者组内，多个消费者实例则会按分区分配实现负载均衡（详见下方make test-multiple-consumers）。

场景 4：不同 offset 类型的消费回放

# 从头开始消费 make consumer OFFSET=earliest # 只消费新消息 make consumer OFFSET=latest # 从指定时间戳（纳秒）开始消费 make consumer OFFSET=timestamp OFFSET_TS=1699000000000000000

earliest适用于 topic 已有存量数据需要回放的场景；timestamp则适合按时间点精确回放，OFFSET_TS必须是纳秒级 Unix 时间戳。

场景 5：多消费者实例负载均衡（内置脚本）

make test-multiple-consumers

该脚本在同一消费者组（multi-consumer-group）下启动两个消费者实例（consumer-1、consumer-2），以multi-testtopic、8 个分区、200 条消息执行，用于验证同一组内多个实例对分区的自动分配与并行消费能力（见 test/mq/Makefile）。

八、常见问题排查

常见错误与对策

开启调试日志

利用 SeaweedFS 的 glog 日志库（仓库 weed/glog）开启 verbose 输出：

GLOG_v=4 make producer GLOG_v=4 make consumer

检查 broker 与 agent 状态

# 检查 broker 集群状态 curl http://localhost:9333/cluster/brokers # 检查 agent 状态（以 server 模式运行时） curl http://localhost:9333/cluster/agents # 或者使用 weed shell 交互式查看 weed shell -master=localhost:9333 > mq.broker.list

九、架构原理与设计要点

从 weed/mq/README.md 的架构说明并结合测试套件的实现，可以梳理出 SeaweedMQ 的核心设计：

1. Agent 中间层架构：测试程序只与 MQ agent 通信（默认 16777 端口），由 agent 作为客户端与 broker 之间的中介，屏蔽了 broker 集群细节；
2. 结构化消息（RecordValue）：消息采用 schema 驱动格式，producer 用StructToSchema反射生成 RecordType，消息以key + RecordValue发布；这一设计使下游可以基于字段过滤（consumer 的-filter参数）和 schema 演进，而不是纯字节流；
3. Topic 与分区管理：创建 topic 时指定分区数（测试默认 4 个分区），分区决定并行度，key 参与分区路由；
4. 消费者组与 offset 管理：ConsumerGroup+ConsumerGroupInstanceId组合定位消费进度，支持 earliest / latest / timestamp 三种重置策略；
5. 负载均衡：同一消费者组内多个实例通过MaxSubscribedPartitions、SlidingWindowSize等参数控制分区订阅与并发消费，实现分区级负载均衡；
6. 容错与弹性：SeaweedMQ 的 broker 无状态且可由 master 动态选取 leader；segment 可随流量自动 split/merge（auto split and merge），broker 异常时可无缝切换到健康 broker 而不丢数据；测试程序的消费者侧也内置了信号优雅退出与后台运行、进程回收的编排。

从更宏观的角度看，SeaweedMQ 的设计目标（"大容量消息接收保存"与"突发流量自动弹性伸缩"）决定了其"计算与存储分离""broker 无状态化""消息按引用传递"等特性——在测试套件中，你能直观观察到这些特性如何通过 agent 会话、segment/分区、消费者组等抽象落地。

十、进阶扩展方向

官方文档建议的下一步工作包括：

1. 修改 producer，通过RecordType发送更复杂的嵌套结构化数据（如 list、record 复合类型）；
2. 在 consumer 中实现基于字段的消息过滤逻辑；
3. 为 producer/consumer 接入指标采集与监控；
4. 部署多个 broker 实例，验证集群弹性与 failover；
5. 进行 schema 演进（schema evolution）测试，验证不同版本 schema 的兼容性。

这些方向也恰好与 weed/mq 目录下broker、sub_coordinator、pub_balancer、segment、offset、schema等子模块的能力相对应，读者可结合对应源码继续深入。

• 分布式文件系统
• 对象存储
• 存储