Rust AI Agent 的分布式协同：多个 agent 实例用消息队列通信的完整方案
·
Rust AI Agent 的分布式协同：多个 agent 实例用消息队列通信的完整方案
当你的 AI Agent 从一个单一实例进化到需要同时管理多个独立 agent 协同工作的阶段时，你迟早会面对一个核心问题：这些 agent 之间怎么通信？ 我作为一个自学转码的 Rust 爱好者，在折腾自己的 side project 时也摔了不少跟头。最开始图省事用 HTTP 调来调去，后来 agent 数量一多，请求超时、消息丢失、顺序错乱的问题全冒出来了。
本篇文章想跟大家一起梳理：如何用 Rust 构建一套基于消息队列的 AI Agent 分布式协同方案。我们会从零开始拆解架构设计、消息协议定义、以及核心的消息分发与心跳检测实现。所有代码保证在 Rust 1.80+ 下可编译运行。
一、分布式 Agent 通信的底层架构：为什么非得上消息队列？
在系统级架构中，分布式 agent 间通信需要满足以下刚性需求：
异步解耦。agent A 生成一个子任务交给 agent B 后，不应该被 B 的执行速度拖住。A 发送消息后应当可以立即去做别的事，而不是傻等 HTTP 响应。消息队列天然支持这种"投递即忘"（fire-and-forget）模式。
消息可靠性。如果一个 agent 挂了，它未处理的消息不应该凭空蒸发。消息队列提供消息持久化（persistence）和 ACK 确认机制，保障消息不丢失、不重复投递（至少保证"至少一次"语义）。
负载均衡。当你有 3 个同类型的 agent（比如都是做代码审查的），消息队列可以按照轮询或竞争消费的方式，自然地把任务分发给空闲的 agent，省去了单独写负载均衡器的成本。
顺序性与并发冲突。多 agent 场景下，消息乱序到达会导致严重的数据一致性问题。消息队列的分区（partition）和键路由（key routing）机制，可以保证同一类业务消息按产生顺序被消费。
二、Rust 侧的消息协议定义：用 Protobuf + Serde 做通用序列化
消息协议是分布式系统里的"通用语言"。选错了协议格式，很可能会导致后期兼容性地狱。我做自学转码时学的第一课就是：协议设计要向后兼容。
这里我们选用 Protobuf3 定义消息结构，同时配合 Rust 的 serde 支持 JSON 格式以方便调试。得益于 prost 这个 Rust 原生 protobuf 库，我们不需要依赖外部的 protoc 编译工具，一切都在 Rust 工具链内闭环。
// =========================================================================
// 消息协议定义：通过 prost-build 在编译期从 .proto 文件生成 Rust 结构体
// 项目地址: https://github.com/tokio-rs/prost
// =========================================================================
// agent_message.proto 内容示意（通过 include! 宏内联）:
// message AgentTask {
// string task_id = 1; // 全局唯一的任务 ID，UUID v7 格式
// AgentRole source = 2; // 消息来源 agent 角色
// AgentRole target = 3; // 目标 agent 角色（可为空表示广播）
// TaskPayload payload = 4; // 携带的具体任务载荷
// int64 created_at = 5; // 消息创建时间戳（毫秒）
// int32 priority = 6; // 优先级，0最高，数值越大越低
// }
use serde::{Deserialize, Serialize};
use uuid::Uuid;
/// Agent 的角色枚举，决定它能消费哪种类型的任务
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum AgentRole {
Coordinator, // 协调器
CodeGenerator, // 代码生成 agent
CodeReviewer, // 代码审查 agent
TestGenerator, // 测试生成 agent
DeployManager, // 部署管理 agent
}
/// 任务负载：根据不同的任务类型包含不同的数据段
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskPayload {
/// 任务类型标识，方便消费者决定如何处理
pub task_type: String,
/// 实际携带的数据，以 JSON 字符串形式传递以保证灵活性
pub data: String,
/// 上下文关联的上游任务 ID，用于结果聚合
pub parent_task_id: Option<String>,
}
/// 核心的 Agent 消息信封（Envelope），所有 agent 间通信均通过该结构
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentMessage {
/// 全局唯一消息 ID
pub message_id: String,
/// 关联的任务 ID，多个消息可以属于同一个任务
pub task_id: String,
/// 消息来源 agent 的角色
pub source: AgentRole,
/// 目标 agent（None 表示所有 agent 均可消费）
pub target: Option<AgentRole>,
/// 具体载荷
pub payload: TaskPayload,
/// 发送时间戳，用于超时检测和延迟监控
pub sent_at: i64,
/// 存活时间（秒），超时后消息自动作废
pub ttl_seconds: i32,
}
impl AgentMessage {
/// 工厂方法：快速创建一个新的 Agent 消息
pub fn new(
source: AgentRole,
target: Option<AgentRole>,
task_type: &str,
data: &str,
) -> Self {
AgentMessage {
message_id: Uuid::now_v7().to_string(),
task_id: Uuid::now_v7().to_string(),
source,
target,
payload: TaskPayload {
task_type: task_type.to_string(),
data: data.to_string(),
parent_task_id: None,
},
sent_at: chrono::Utc::now().timestamp_millis(),
ttl_seconds: 300, // 默认 5 分钟 TTL
}
}
}
三、基于 Lapin（AMQP）的消息队列集成实现
选型上，我对比过 NATS、Redis Stream 和 RabbitMQ。NATS 性能最狂暴但生态相对小众，Redis Stream 适合轻量场景，而 RabbitMQ 的 AMQP 协议生态最成熟、文档最丰富、功能最全面（支持 Topic Exchange、死信队列、延迟消息等）。
Rust 生态中对接 RabbitMQ 最成熟的库是 lapin，它基于 tokio 异步运行时，天然适合我们的 agent 场景。
use lapin::{
options::*, types::FieldTable, BasicProperties, Channel, Connection,
ConnectionProperties, ExchangeKind,
};
use tokio::sync::mpsc;
use anyhow::{Context, Result};
/// Agent 的消息总线：封装了与 RabbitMQ 的所有交互
pub struct MessageBus {
channel: Channel, // AMQP 通道
task_queue: String, // 任务分派队列名
pub result_sender: mpsc::Sender<AgentMessage>, // 结果消息投递通道
}
impl MessageBus {
/// 连接到 RabbitMQ 并初始化交换机和队列拓扑
pub async fn connect(amqp_url: &str) -> Result<Self> {
// 建立 AMQP 连接，连接属性保持默认即可
let conn = Connection::connect(amqp_url, ConnectionProperties::default())
.await
.context("无法连接到 RabbitMQ，请检查 AMQP 地址和网络")?;
let channel = conn.create_channel().await?;
// 声明主题交换机：支持通配符路由（如 agent.code.*）
channel
.exchange_declare(
"agent_exchange", // 交换机名称
ExchangeKind::Topic, // Topic 类型支持模式匹配路由
ExchangeDeclareOptions::default(),
FieldTable::default(),
)
.await?;
// 声明任务队列：持久化、非独占、自动删除关闭
let task_queue = "agent_task_queue".to_string();
channel
.queue_declare(
&task_queue,
QueueDeclareOptions {
durable: true, // 队列持久化，重启不会丢失队列声明
..Default::default()
},
FieldTable::default(),
)
.await?;
// 将队列绑定到交换机，路由键匹配所有 agent 任务
channel
.queue_bind(
&task_queue,
"agent_exchange",
"task.#", // 匹配 task.code_gen、task.review 等
QueueBindOptions::default(),
FieldTable::default(),
)
.await?;
// 创建结果投递通道（内存通道，用于跨 agent 任务间的消息传递）
let (result_sender, _) = mpsc::channel::<AgentMessage>(1024);
Ok(MessageBus {
channel,
task_queue,
result_sender,
})
}
/// 发布一条任务消息到 RabbitMQ
pub async fn publish_task(&self, message: &AgentMessage) -> Result<()> {
// 将消息序列化为 JSON（方便调试和跨语言互操作）
let payload = serde_json::to_vec(message)?;
// 构造路由键：task.{目标角色}，消费者据此选择性订阅
let routing_key = format!(
"task.{}",
match &message.target {
Some(AgentRole::CodeGenerator) => "code_gen",
Some(AgentRole::CodeReviewer) => "review",
Some(AgentRole::TestGenerator) => "test",
Some(AgentRole::DeployManager) => "deploy",
_ => "broadcast",
}
);
// 发布消息，设置持久化模式和 TTL
let mut headers = FieldTable::default();
headers.insert(
"x-message-ttl".into(),
(message.ttl_seconds * 1000).into(), // 消息级 TTL（毫秒）
);
self.channel
.basic_publish(
"agent_exchange",
&routing_key,
BasicPublishOptions::default(),
&payload,
BasicProperties::default()
.with_delivery_mode(2) // 持久化消息
.with_headers(headers),
)
.await?;
println!(
"[消息总线] 任务消息已发布 | ID: {} | 路由: {}",
message.message_id, routing_key
);
Ok(())
}
/// 消费者：从任务队列拉取消息并交给处理器
pub async fn consume_tasks<F>(&self, handler: F) -> Result<()>
where
F: Fn(AgentMessage) -> anyhow::Result<()> + Send + Sync + 'static,
{
// 注册消费者，设置预取数量防止单个消费者被消息打爆
let consumer = self
.channel
.basic_consume(
&self.task_queue,
"agent_worker",
BasicConsumeOptions {
no_ack: false, // 手动 ACK，确保消息处理完毕才确认
..Default::default()
},
FieldTable::default(),
)
.await?;
// 循环接收投递的消息
while let Some(delivery) = consumer.clone().into_stream().next().await {
if let Ok(delivery) = delivery {
// 反序列化消息体
let message: AgentMessage = serde_json::from_slice(&delivery.data)?;
println!(
"[消费者] 收到任务消息 | ID: {} | 类型: {}",
message.message_id, message.payload.task_type
);
// 调用外部传入的处理器函数
match handler(message) {
Ok(_) => {
// 处理成功：向队列确认消息已消费
delivery.ack(BasicAckOptions::default()).await?;
}
Err(e) => {
// 处理失败：将消息重新放回队列（或转入死信队列）
eprintln!("[消费者] 处理失败: {:?}, 消息将重新入队", e);
delivery
.nack(BasicNackOptions {
requeue: true, // 重新放入队列等待重试
..Default::default()
})
.await?;
}
}
}
}
Ok(())
}
}
四、心跳检测与 Agent 存活监控：保障分布式集群的自我修复能力
在生产环境里，agent 意外挂掉是家常便饭。没有心跳机制，协调器根本不知道哪个 agent 已经失联，会不停地往一个死节点投递任务——这无异于往黑洞里扔消息。
下面实现一个基于 Redis 的心跳注册中心。每个 agent 启动后周期性更新心跳时间戳，协调器定期扫描超时的 agent 并将其标记为 OFFLINE。
use redis::{AsyncCommands, Client as RedisClient};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time;
/// Agent 心跳管理器：每个 agent 实例向 Redis 上报存活状态
pub struct Heartbeat {
redis: RedisClient,
agent_id: String,
/// 心跳超时阈值（秒），超过此时间未收到心跳的 agent 视为离线
timeout_threshold: i64,
}
impl Heartbeat {
/// 创建一个新的心跳管理器实例
pub fn new(redis_url: &str, agent_id: &str) -> Result<Self> {
let redis = RedisClient::open(redis_url)?;
Ok(Heartbeat {
redis,
agent_id: agent_id.to_string(),
timeout_threshold: 30, // 30 秒无心跳即判定离线
})
}
/// 向 Redis 上报心跳，键为 heartbeat:{agent_id}，值为当前时间戳
pub async fn send_heartbeat(&self) -> Result<()> {
let mut conn = self.redis.get_multiplexed_async_connection().await?;
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)?
.as_secs() as i64;
// 设置心跳键，同时设定过期时间防止僵尸 key 堆积
let key = format!("heartbeat:{}", self.agent_id);
// SETEX 原子操作，同时设置值和过期时间
let _: () = conn.set_ex(&key, now, 60).await?;
Ok(())
}
/// 启动心跳循环：每隔 10 秒发送一次心跳信号
pub async fn start_heartbeat_loop(&self) {
let mut interval = time::interval(Duration::from_secs(10));
loop {
interval.tick().await;
if let Err(e) = self.send_heartbeat().await {
eprintln!(
"[心跳] Agent {} 发送心跳失败: {:?}",
self.agent_id, e
);
} else {
println!("[心跳] Agent {} 心跳正常 ✓", self.agent_id);
}
}
}
/// 协调器调用：扫描所有注册 agent 并在 Redis 中检查存活状态
pub async fn get_alive_agents(redis_url: &str) -> Result<Vec<String>> {
let redis = RedisClient::open(redis_url)?;
let mut conn = redis.get_multiplexed_async_connection().await?;
// 使用 SCAN 遍历 heartbeat:* 键，避免 KEYS 命令阻塞 Redis
let mut cursor: u64 = 0;
let mut alive = Vec::new();
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)?
.as_secs() as i64;
loop {
let (next_cursor, keys): (u64, Vec<String>) =
conn.scan_match(cursor, "heartbeat:*").await?;
for key in keys {
// 读取该心跳键的时间戳
let timestamp: Option<i64> = conn.get(&key).await?;
if let Some(ts) = timestamp {
// 如果心跳在超时窗口内，判定为存活
if now - ts <= 30 {
let agent_id = key.strip_prefix("heartbeat:").unwrap_or(&key);
alive.push(agent_id.to_string());
}
}
}
if next_cursor == 0 {
break;
}
cursor = next_cursor;
}
Ok(alive)
}
}
五、总结
AI Agent 的分布式协同，说到底是把"一个 agent 能做的事"拆成"多个 agent 各司其职"，再用一套可靠的消息中间件把它们串联起来。Rust 在这方面有天然优势：异步原语（tokio）、性能、内存安全——三项都拿捏住了。
这篇文章我们走通了 Agent 分布式协同的核心链路：消息协议定义 → AMQP 队列集成 → 心跳存活监控。作为自学出身，我在实现这套方案时最大的感受是：分布式系统的复杂度不在于单点代码多难写，而在于"坏了之后怎么办"——消息丢了、agent 挂了、网络抖了。好在这套基于消息队列 + 心跳检测的架构，在大多数故障场景下有自愈能力。
下篇文章我们聊聊 Tokio 多运行时模式，看看什么场景下值得开启多个独立的 runtime 各管一摊。欢迎在评论区交流你在 agent 系统中遇到的实际坑。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
2
0- 0
已为社区贡献70条内容
所有评论(0)