Agent Client Protocol 实战：构建高可靠分布式系统的通信框架

1次阅读

共计 2709 个字符，预计需要花费 7 分钟才能阅读完成。

在分布式系统中，Agent（代理）与 Client（客户端）的通信质量直接影响系统可靠性。传统 HTTP/REST 架构在此场景下暴露三个核心问题：

长连接维护成本高 ：频繁建立 TCP 连接导致延迟飙升，尤其在高并发场景下
双向流支持不足 ：HTTP/1.x 的请求 - 响应模式难以实现服务端主动推送（如实时监控数据）
跨语言兼容性差 ：不同语言实现的 JSON 序列化 / 反序列化常有细微差异

对比主流通信方案：

方案	长连接	双向流	编码效率	多语言支持
HTTP/1.1	❌	❌	中	✅
WebSocket	✅	✅	低	✅
RSocket	✅	✅	高	⚠️
gRPC	✅	✅	高	✅

gRPC 基于 HTTP/ 2 的特性带来三大优势：

Protobuf 二进制编码 ：比 JSON 体积小 3 -10 倍
原生多语言 SDK：自动生成 Go/Python/Java 等代码
四种通信模式 ：支持一元调用、服务端流、客户端流、双向流

syntax = "proto3";

// 元数据头（Metadata Header）message Metadata {
  string request_id = 1;  // 请求唯一 ID
  int64 timestamp = 2;    // 毫秒级时间戳
  map<string, string> headers = 3;
}

// 定义 Agent 服务
service AgentService {
  // 简单 RPC（一元调用）rpc ExecuteCommand (CommandRequest) returns (CommandResponse);

  // 服务端流式 RPC
  rpc StreamLogs (LogRequest) returns (stream LogChunk);

  // 双向流式 RPC 
  rpc Chat (stream ChatMessage) returns (stream ChatMessage);
}

message CommandRequest {
  Metadata meta = 1;
  string command = 2;
  int32 timeout_sec = 3;  // 超时控制
}

请求 ID（request_id）：采用 UUIDv4 实现分布式追踪
超时传递（timeout_sec）：客户端显式传递超时时间，服务端需遵守
元数据头（headers）：传递鉴权 token、路由标签等业务信息

// 实现连接池
type AgentPool struct {pool sync.Pool}

func (p *AgentPool) Get() *grpc.ClientConn {conn := p.pool.Get()
  if conn == nil {
    conn, _ = grpc.Dial("localhost:50051", 
      grpc.WithKeepaliveParams(keepalive.ClientParameters{Time: 30 * time.Second, // 心跳间隔}))
  }
  return conn.(*grpc.ClientConn)
}

// 心跳协程
func startHeartbeat(stream pb.AgentService_ChatServer) {ticker := time.NewTicker(10 * time.Second)
  for {
    select {
    case <-ticker.C:
      stream.Send(&pb.ChatMessage{Type: "PING"})
    case <-stream.Context().Done():
      return
    }
  }
}

# 带重试的 Client 封装
class RetryClient:
    def __init__(self, max_retries=3):
        self.channel = grpc.insecure_channel(
            'localhost:50051',
            options=[('grpc.enable_retries', 1),
                ('grpc.service_config', 
                 '{"retryPolicy": {"maxAttempts": 5}}')
            ])

    def execute_command(self, cmd):
        stub = pb.AgentServiceStub(self.channel)
        try:
            return stub.ExecuteCommand(pb.CommandRequest(
                command=cmd,
                timeout_sec=10
            ), timeout=15)  # 客户端总超时 > 服务端超时
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
                logging.error("Timeout when executing command")
            raise

# gRPC 服务端配置
server:
  grpc:
    keepalive:
      time: 1h           # 空闲连接存活时间
      timeout: 20s       # ping 超时
      min_time: 30s      # 最小 ping 间隔

算法	压缩率	压缩耗时 (ms)	解压耗时 (ms)
None	1.0x	0	0
Gzip	3.2x	12	8
Zstd	3.8x	9	5

触发条件：客户端发送 ping 的频率高于服务端配置的 min_time

解决方案：

统一客户端与服务端的 keepalive 参数
服务端添加拦截器统计 ping 频率

func rateLimitInterceptor(ctx context.Context) {if atomic.LoadInt32(&pingCount) > 10 {
    return status.Error(
      codes.ResourceExhausted, 
      "too many pings")
  }
}

预留字段号段：
1-15：高频使用字段（占用 1 字节）
16-2047：常规字段
不要修改已上线字段的编号

实现平滑升级的三种方案：

字段规则 ：
新字段使用新编号
废弃字段保留编号但标记 reserved
协议协商 ：
客户端在首次握手时发送支持的最高版本号
服务端选择双方兼容的最高版本
双跑策略 ：
新老版本服务并行运行
通过流量灰度逐步迁移

经过实际压测，这套基于 gRPC 的协议在 1k 字节消息体下达到 15 万 QPS，平均延迟仅 2.3ms。最关键的是通过良好的协议设计，使业务逻辑与通信解耦，后续扩展新的 RPC 方法时，团队成员能快速理解并实现。建议在正式环境部署时，配合 Prometheus 监控 gRPC 的指标（如 grpc_server_handled_total）。

正文完