📖 第18课：四种通信模式

阶段四：gRPC

gRPC的四种通信模式——一元RPC、服务端流、客户端流、双向流——是它区别于REST的核心优势。本课深入每种模式的原理、适用场景和完整实现，让你能根据业务需求选择最合适的通信模式。

📊 四种模式速览

模式	Proto语法	请求	响应	典型场景
一元RPC	`rpc Method(Req) returns (Res)`	1个	1个	CRUD、查询
服务端流	`rpc Method(Req) returns (stream Res)`	1个	流	导出、日志、监控
客户端流	`rpc Method(stream Req) returns (Res)`	流	1个	上传、批量导入
双向流	`rpc Method(stream Req) returns (stream Res)`	流	流	聊天、协作、代理

1️⃣ 一元RPC（Unary RPC）

// Proto定义
service OrderService {
  rpc GetOrder(GetOrderRequest) returns (Order);
  rpc CreateOrder(CreateOrderRequest) returns (Order);
}

// 服务端实现
function GetOrder(call, callback) {
  const order = orders.find(o => o.id === call.request.id);
  if (!order) {
    return callback({
      code: grpc.status.NOT_FOUND,
      message: `订单 ${call.request.id} 不存在`,
      details: Buffer.from(JSON.stringify({ orderId: call.request.id })),
    });
  }
  callback(null, order);
}

function CreateOrder(call, callback) {
  const req = call.request;
  // 验证
  if (!req.userId) {
    return callback({
      code: grpc.status.INVALID_ARGUMENT,
      message: 'userId不能为空',
    });
  }
  if (!req.items || req.items.length === 0) {
    return callback({
      code: grpc.status.INVALID_ARGUMENT,
      message: '订单必须包含至少一个商品',
    });
  }
  // 创建
  const order = { id: nextId++, ...req, status: 'PENDING', createdAt: new Date().toISOString() };
  orders.push(order);
  callback(null, order);
}

// 客户端调用
client.GetOrder({ id: 1 }, (err, order) => {
  if (err) {
    console.error(`错误[${err.code}]: ${err.message}`);
    return;
  }
  console.log('订单:', order);
});

// 使用Promise包装
function getOrderAsync(id) {
  return new Promise((resolve, reject) => {
    client.GetOrder({ id }, (err, order) => {
      if (err) reject(err);
      else resolve(order);
    });
  });
}

2️⃣ 服务端流（Server Streaming）

// Proto定义
service AnalyticsService {
  // 实时导出订单数据
  rpc ExportOrders(ExportRequest) returns (stream Order);
  // 实时监控指标
  rpc StreamMetrics(MetricsRequest) returns (stream Metric);
  // 搜索结果分批返回
  rpc Search(SearchRequest) returns (stream SearchResult);
}

// 服务端实现：逐条发送
function ExportOrders(call) {
  const { status, startDate, endDate } = call.request;

  let filtered = [...orders];
  if (status) filtered = filtered.filter(o => o.status === status);

  // 逐条发送，内存友好
  let index = 0;
  const batchSize = 10;

  function sendBatch() {
    const end = Math.min(index + batchSize, filtered.length);
    for (; index < end; index++) {
      call.write(filtered[index]);
    }

    if (index < filtered.length) {
      // 继续发送下一批
      setImmediate(sendBatch);
    } else {
      call.end();
    }
  }

  sendBatch();
}

// 实时指标推送
function StreamMetrics(call) {
  const interval = setInterval(() => {
    const metric = {
      timestamp: Date.now(),
      cpuUsage: Math.random() * 100,
      memoryUsage: Math.random() * 100,
      requestRate: Math.floor(Math.random() * 1000),
      errorRate: Math.random() * 5,
    };
    call.write(metric);
  }, 1000);

  // 客户端取消时清理
  call.on('cancelled', () => {
    clearInterval(interval);
    console.log('客户端取消订阅');
  });
}

// 客户端接收
const stream = client.ExportOrders({ status: 'SHIPPED' });
stream.on('data', (order) => {
  console.log(`导出订单: ${order.orderNumber}`);
});
stream.on('end', () => console.log('导出完成'));
stream.on('error', (err) => console.error('流错误:', err));

// 监控流
const metricsStream = client.StreamMetrics({});
metricsStream.on('data', (metric) => {
  process.stdout.write(`\rCPU: ${metric.cpuUsage.toFixed(1)}% | MEM: ${metric.memoryUsage.toFixed(1)}% | RPS: ${metric.requestRate}`);
});

3️⃣ 客户端流（Client Streaming）

// Proto定义
service ImportService {
  rpc ImportProducts(stream Product) returns (ImportResponse);
  rpc UploadFile(stream FileChunk) returns (UploadResponse);
}

// 服务端实现：收集所有数据后处理
function ImportProducts(call, callback) {
  const products = [];
  const errors = [];

  call.on('data', (product) => {
    // 逐条验证
    if (!product.name) {
      errors.push(`第${products.length + 1}条: name为空`);
      return;
    }
    products.push({ ...product, id: nextId++ });
  });

  call.on('end', () => {
    // 批量写入
    products.forEach(p => allProducts.push(p));
    callback(null, {
      successCount: products.length,
      failCount: errors.length,
      errors,
    });
  });
}

// 大文件上传
function UploadFile(call, callback) {
  const chunks = [];
  let totalSize = 0;
  let fileName = '';

  call.on('data', (chunk) => {
    if (!fileName && chunk.fileName) fileName = chunk.fileName;
    chunks.push(chunk.data);
    totalSize += chunk.data.length;
    console.log(`已接收: ${(totalSize / 1024 / 1024).toFixed(2)} MB`);
  });

  call.on('end', () => {
    const fileId = `file_${Date.now()}`;
    // 实际写入磁盘或对象存储
    console.log(`文件 ${fileName} 上传完成，共 ${(totalSize / 1024 / 1024).toFixed(2)} MB`);
    callback(null, { fileId, fileName, size: totalSize, url: `/files/${fileId}` });
  });
}

// 客户端发送流
function importProducts(products) {
  const stream = client.ImportProducts((err, response) => {
    if (err) return console.error('导入失败:', err);
    console.log(`导入完成: 成功${response.successCount}, 失败${response.failCount}`);
  });

  products.forEach(product => stream.write(product));
  stream.end();
}

// 大文件分块上传
function uploadFile(filePath) {
  const fs = require('fs');
  const stream = client.UploadFile((err, response) => {
    if (err) return console.error('上传失败:', err);
    console.log(`上传成功: ${response.url}`);
  });

  const CHUNK_SIZE = 64 * 1024; // 64KB
  const buffer = fs.readFileSync(filePath);
  const fileName = require('path').basename(filePath);

  for (let offset = 0; offset < buffer.length; offset += CHUNK_SIZE) {
    const end = Math.min(offset + CHUNK_SIZE, buffer.length);
    stream.write({
      fileName: offset === 0 ? fileName : '',
      data: buffer.slice(offset, end),
    });
  }
  stream.end();
}

4️⃣ 双向流（Bidirectional Streaming）

// Proto定义
service ChatService {
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
  rpc Proxy(stream ProxyRequest) returns (stream ProxyResponse);
}

// 服务端实现：聊天室
const chatRooms = new Map(); // roomId → Set of call objects

function Chat(call) {
  let currentUser = null;
  let currentRoom = null;

  call.on('data', (msg) => {
    // 首次消息：加入房间
    if (!currentUser) {
      currentUser = msg.user;
      currentRoom = msg.roomId || 'default';
      if (!chatRooms.has(currentRoom)) chatRooms.set(currentRoom, new Set());
      chatRooms.get(currentRoom).add(call);
      // 广播加入消息
      broadcast(currentRoom, { user: '系统', message: `${currentUser} 加入了聊天`, roomId: currentRoom });
      return;
    }

    // 广播消息到房间所有人
    broadcast(currentRoom, msg);
  });

  call.on('end', () => {
    if (currentRoom && chatRooms.has(currentRoom)) {
      chatRooms.get(currentRoom).delete(call);
      broadcast(currentRoom, { user: '系统', message: `${currentUser} 离开了聊天`, roomId: currentRoom });
    }
  });
}

function broadcast(roomId, message) {
  const room = chatRooms.get(roomId);
  if (!room) return;
  room.forEach(client => {
    try { client.write(message); } catch (e) { /* 客户端已断开 */ }
  });
}

// 客户端：聊天
function startChat(userName) {
  const stream = client.Chat();

  stream.on('data', (msg) => {
    console.log(`[${msg.user}]: ${msg.message}`);
  });

  // 发送加入消息
  stream.write({ user: userName, message: '', roomId: 'general' });

  // 从stdin读取输入
  const readline = require('readline');
  const rl = readline.createInterface({ input: process.stdin });
  rl.on('line', (line) => {
    stream.write({ user: userName, message: line, roomId: 'general' });
  });
}

🛠️ 综合实战：监控平台

// monitoring.proto
syntax = "proto3";
package monitoring;

service MonitoringService {
  // 一元：查询单个指标
  rpc GetMetric(GetMetricRequest) returns (Metric);
  
  // 服务端流：实时监控面板
  rpc StreamDashboard(DashboardRequest) returns (stream DashboardUpdate);
  
  // 客户端流：批量上报指标
  rpc ReportMetrics(stream MetricReport) returns (ReportResponse);
  
  // 双向流：交互式调试
  rpc DebugSession(stream DebugCommand) returns (stream DebugOutput);
}

message Metric { string name = 1; double value = 2; int64 timestamp = 3; }
message GetMetricRequest { string name = 1; }
message DashboardRequest { repeated string metrics = 1; int32 interval_ms = 2; }
message DashboardUpdate { repeated Metric metrics = 1; int64 timestamp = 2; }
message MetricReport { string name = 1; double value = 2; string source = 3; }
message ReportResponse { int32 accepted = 1; int32 rejected = 2; }
message DebugCommand { string command = 1; map<string,string> args = 2; }
message DebugOutput { string output = 1; bool is_error = 2; }

📝 本课小结

一元RPC适合CRUD操作，最常用最简单
服务端流适合导出、监控、日志等服务端推送场景
客户端流适合上传、批量导入等客户端推送场景
双向流适合聊天、协作、代理等交互式场景
流式调用注意资源清理（interval/call.on('cancelled')）
大文件上传使用客户端流分块发送

📐 流式设计最佳实践

流式API设计清单

分块大小：每条消息控制在几KB以内，避免单条过大
进度反馈：长流式操作定期发送进度消息
取消支持：服务端检查call.cancelled，客户端可随时取消
超时设置：流式调用设置合理的deadline
错误恢复：客户端记录最后成功ID，支持断点续传
背压处理：生产端感知消费端速度，避免内存溢出

// 流式API设计模板

// 1. 带进度的服务端流
rpc ExportData(ExportRequest) returns (stream ExportChunk);

message ExportChunk {
  oneof payload {
    DataRecord record = 1;      // 数据记录
    ProgressUpdate progress = 2; // 进度更新
    ExportError error = 3;       // 错误信息
  }
  int64 sequence = 4;           // 序列号
  string cursor = 5;            // 断点续传游标
}

// 2. 带确认的客户端流
rpc UploadData(stream UploadChunk) returns (stream UploadAck);

message UploadChunk {
  bytes data = 1;
  int64 sequence = 2;
  bool is_last = 3;
}

message UploadAck {
  int64 received_sequence = 1;
  bool success = 2;
  string error = 3;
}

💪 练习

练习1：实现日志收集服务

设计一个日志收集gRPC服务：应用通过客户端流上报日志，监控系统通过服务端流订阅实时日志。

练习2：实现gRPC代理

用双向流实现一个TCP代理：客户端发送请求，代理转发到后端服务，后端响应通过代理回传。

🏆 本课成就：流式通信大师

✅ 掌握四种gRPC通信模式的实现
✅ 理解每种模式的适用场景
✅ 实现服务端流导出和监控
✅ 实现客户端流上传和导入
✅ 实现双向流聊天和代理