📖 第19课：流式处理

阶段四：gRPC

流式处理是gRPC的高级主题——背压控制、流控窗口、超时管理、错误恢复、流式中间件。这些机制让gRPC在生产环境中可靠运行，处理海量数据而不崩溃。本课深入流式处理的工程实践。

🌊 背压与流控

背压(Backpressure)是流式系统的核心机制：当下游处理速度慢于上游发送速度时，通过反馈信号让上游减速，避免内存溢出。

// gRPC流控实现
// HTTP/2内置流控窗口，gRPC自动处理

// 服务端：感知客户端的消费速度
function StreamLargeDataset(call) {
  let paused = false;
  let index = 0;
  const totalRecords = 1000000; // 100万条数据

  // 监听客户端的流控信号
  call.on('pause', () => {
    paused = true;
    console.log('⏸ 客户端请求暂停');
  });

  call.on('resume', () => {
    paused = false;
    console.log('▶ 客户端恢复接收');
    sendNext();
  });

  function sendNext() {
    if (paused || index >= totalRecords) {
      if (index >= totalRecords) call.end();
      return;
    }

    // 每批发送100条
    const batchSize = 100;
    for (let i = 0; i < batchSize && index < totalRecords; i++, index++) {
      call.write({
        id: index,
        data: generateRecord(index),
        progress: Math.round(index / totalRecords * 100),
      });
    }

    // 让出事件循环，允许流控信号处理
    setImmediate(sendNext);
  }

  sendNext();
}

// 客户端：控制接收速度
function receiveLargeDataset() {
  const stream = client.StreamLargeDataset({});
  let received = 0;
  const buffer = [];

  stream.on('data', (record) => {
    received++;
    buffer.push(record);

    // 模拟慢速处理
    processRecord(record);

    // 当缓冲区超过阈值时暂停
    if (buffer.length > 1000) {
      stream.pause();  // ⏸ 告诉服务端暂停发送
      console.log(`⏸ 暂停接收，缓冲区: ${buffer.length}`);
      
      // 处理完缓冲区后恢复
      setTimeout(() => {
        buffer.length = 0;
        stream.resume();  // ▶ 恢复接收
        console.log('▶ 恢复接收');
      }, 1000);
    }
  });

  stream.on('end', () => console.log(`接收完成: ${received}条`));
}

function processRecord(record) {
  // 模拟耗时处理
  for (let i = 0; i < 100; i++) Math.sqrt(i);
}

⏱️ 超时与截止时间

// Deadline：截止时间（推荐） vs Timeout：超时时间
// Deadline是绝对时间点，Timeout是相对时长
// 跨服务调用时Deadline自动传播，Timeout不会

// 客户端设置Deadline
const deadline = new Date();
deadline.setSeconds(deadline.getSeconds() + 5); // 5秒截止

client.GetOrder({ id: 1 }, { deadline }, (err, order) => {
  if (err) {
    if (err.code === grpc.status.DEADLINE_EXCEEDED) {
      console.error('请求超时');
    }
    return;
  }
  console.log(order);
});

// 服务端检查Deadline
function ProcessOrder(call, callback) {
  // 定期检查是否还有时间
  if (call.cancelled) {
    return callback({ code: grpc.status.CANCELLED, message: '请求已取消' });
  }

  // 长时间操作中检查deadline
  function longOperation(step) {
    if (call.cancelled) {
      console.log('客户端已取消，停止处理');
      return;
    }

    // 处理当前步骤
    const result = doStep(step);

    if (step < totalSteps) {
      // 让出事件循环，允许检查取消信号
      setImmediate(() => longOperation(step + 1));
    } else {
      callback(null, result);
    }
  }

  longOperation(1);
}

// 超时重试策略
async function getOrderWithRetry(id, maxRetries = 3) {
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    const timeout = Math.min(1000 * Math.pow(2, attempt), 10000); // 指数退避
    const deadline = new Date(Date.now() + timeout);
    
    try {
      return await new Promise((resolve, reject) => {
        client.GetOrder({ id }, { deadline }, (err, order) => {
          if (err) reject(err);
          else resolve(order);
        });
      });
    } catch (err) {
      if (err.code === grpc.status.DEADLINE_EXCEEDED && attempt < maxRetries) {
        console.log(`第${attempt}次重试...`);
        await new Promise(r => setTimeout(r, 100 * attempt));
        continue;
      }
      throw err;
    }
  }
}

🔄 流式错误恢复

// 流式调用的错误恢复模式

// 1. 可恢复的服务端流
function ResilientServerStream(call) {
  let lastSentId = 0;
  
  // 客户端可以发送最后收到的ID来恢复
  const resumeFromId = call.request.resumeFromId || 0;

  const dataStream = getDataStream(resumeFromId);

  dataStream.on('data', (record) => {
    try {
      call.write(record);
      lastSentId = record.id;
    } catch (e) {
      console.error('写入失败，客户端可能已断开');
      dataStream.destroy();
    }
  });

  dataStream.on('end', () => call.end());
  dataStream.on('error', (err) => {
    // 发送错误状态但不断开流
    call.write({
      id: -1,
      error: { code: 'STREAM_ERROR', message: err.message, lastSuccessfulId: lastSentId },
    });
    // 尝试重新连接数据源
    setTimeout(() => reconnectDataStream(lastSentId), 1000);
  });
}

// 2. 客户端断线重连
class ResilientStreamClient {
  constructor(client, options = {}) {
    this.client = client;
    this.maxRetries = options.maxRetries || 5;
    this.retryDelay = options.retryDelay || 1000;
    this.lastReceivedId = 0;
  }

  async subscribe(request) {
    let attempt = 0;
    
    while (attempt < this.maxRetries) {
      try {
        await new Promise((resolve, reject) => {
          const stream = this.client.StreamData({
            ...request,
            resumeFromId: this.lastReceivedId,  // 断点续传
          });

          stream.on('data', (data) => {
            if (data.error) {
              console.warn(`流错误: ${data.error.message}，最后ID: ${data.error.lastSuccessfulId}`);
              return;
            }
            this.lastReceivedId = data.id;
            this.onData(data);
          });

          stream.on('end', resolve);
          stream.on('error', reject);
        });

        // 正常结束
        return;
      } catch (err) {
        attempt++;
        console.error(`流断开(第${attempt}次)，${this.retryDelay * attempt}ms后重连...`);
        await new Promise(r => setTimeout(r, this.retryDelay * attempt));
      }
    }

    throw new Error('重连次数超限');
  }

  onData(data) {
    // 子类覆盖
    console.log('收到数据:', data.id);
  }
}

🛠️ 流式中间件

// gRPC流式中间件模式

// 日志中间件
function loggingMiddleware(handler) {
  return function(call) {
    const start = Date.now();
    const methodName = call.call?.method || 'unknown';
    
    console.log(`→ ${methodName} 开始`);
    
    if (call.write) {
      // 一元RPC
      const originalCallback = arguments[1];
      return handler(call, (err, response) => {
        const duration = Date.now() - start;
        if (err) {
          console.log(`← ${methodName} 错误 [${err.code}] (${duration}ms)`);
        } else {
          console.log(`← ${methodName} 成功 (${duration}ms)`);
        }
        originalCallback?.(err, response);
      });
    }

    // 流式RPC
    const originalWrite = call.write.bind(call);
    let messageCount = 0;
    call.write = (data) => {
      messageCount++;
      return originalWrite(data);
    };

    call.on('end', () => {
      const duration = Date.now() - start;
      console.log(`← ${methodName} 流结束 (${messageCount}条, ${duration}ms)`);
    });

    return handler(call);
  };
}

// 限流中间件
function rateLimitMiddleware(maxConcurrent = 100) {
  let activeStreams = 0;
  
  return function(handler) {
    return function(call) {
      if (activeStreams >= maxConcurrent) {
        if (call.callback) {
          call.callback({ code: grpc.status.RESOURCE_EXHAUSTED, message: '并发流超限' });
        } else {
          call.emit('error', { code: grpc.status.RESOURCE_EXHAUSTED, message: '并发流超限' });
        }
        return;
      }
      
      activeStreams++;
      call.on('end', () => { activeStreams--; });
      call.on('error', () => { activeStreams--; });
      
      return handler(call);
    };
  };
}

// 指标收集中间件
function metricsMiddleware(handler) {
  return function(call) {
    const startTime = process.hrtime.bigint();
    const method = call.call?.method || 'unknown';
    
    const endHandler = () => {
      const duration = Number(process.hrtime.bigint() - startTime) / 1e6;
      metrics.record(method, duration);
    };
    
    call.on('end', endHandler);
    call.on('error', endHandler);
    
    return handler(call);
  };
}

📝 本课小结

背压机制防止生产者压垮消费者——pause/resume控制流量
Deadline优于Timeout——跨服务自动传播
流式调用需要断线重连和断点续传策略
指数退避重试避免雪崩
流式中间件实现日志、限流、指标等横切关注点
长时间操作定期检查cancelled状态

📋 流式处理检查清单

生产级流式处理检查项

检查项	说明	默认值
Deadline设置	所有流式调用设置截止时间	30s-5min
背压处理	生产端感知消费端速度	必选
取消检查	服务端定期检查cancelled	每100ms
重试策略	断点续传+指数退避	最多3次
消息大小限制	单条消息不超过4MB	4MB
心跳/保活	长时间流需要心跳	30s间隔
资源清理	流结束时释放资源	必选
监控指标	消息数、延迟、错误率	必选

// 生产级流式服务模板
const grpc = require('@grpc/grpc-js');

// 心跳保活
function createHeartbeat(call, intervalMs = 30000) {
  return setInterval(() => {
    if (call.cancelled) return clearInterval(this);
    try {
      call.write({ type: 'HEARTBEAT', timestamp: Date.now() });
    } catch (e) {
      clearInterval(this);
      console.log('心跳失败，客户端可能已断开');
    }
  }, intervalMs);
}

// 带完整保障的服务端流
function productionStream(call) {
  const heartbeat = createHeartbeat(call);
  let sequence = 0;
  let lastCursor = null;

  call.on('cancelled', () => {
    clearInterval(heartbeat);
    console.log(`流被取消 [seq=${sequence}, cursor=${lastCursor}]`);
  });

  function sendData(data) {
    if (call.cancelled) return;
    sequence++;
    lastCursor = data.id;
    call.write({ ...data, sequence, cursor: lastCursor });
  }

  // ...业务逻辑
}

// gRPC流式错误码
const GRPC_STREAM_ERRORS = {
  DEADLINE_EXCEEDED: { httpCode: 504, message: '处理超时' },
  RESOURCE_EXHAUSTED: { httpCode: 429, message: '资源耗尽' },
  UNAVAILABLE: { httpCode: 503, message: '服务不可用' },
  INTERNAL: { httpCode: 500, message: '内部错误' },
};

// 流式指标收集
const streamMetrics = {
  activeStreams: 0,
  totalMessages: 0,
  errors: 0,
  avgLatency: 0,
  recordMessage() { this.totalMessages++; },
  recordError() { this.errors++; },
  getReport() { return { ...this, errorRate: this.errors / Math.max(1, this.totalMessages) }; },
};

💪 练习

练习1：实现可靠的文件传输

用客户端流实现大文件上传，支持：分块传输、校验和验证、断点续传、进度报告。

练习2：实现流式限流

限制每个IP的并发流数量不超过5个，超过返回RESOURCE_EXHAUSTED错误。

🏆 本课成就：流式处理专家

✅ 掌握背压与流控机制
✅ 实现Deadline和超时重试
✅ 实现流式错误恢复和断线重连
✅ 实现流式中间件模式