Python 与 Node 里读 SSE 流：别让连接泄漏

为什么流读取容易泄漏连接

SSE（Server-Sent Events）是一种基于长连接的流式传输协议。当客户端发起请求后，服务器会保持连接并持续推送数据。如果客户端在流结束前主动断开（例如用户取消生成、超时、程序异常），而代码没有正确关闭底层连接，就会导致连接泄漏。

泄漏的连接会占用文件描述符、内存和服务器资源，长期运行的服务可能因此变慢甚至崩溃。

OpenAI SDK 与原生 HTTP 的对比

使用 OpenAI SDK（推荐）

OpenAI 官方的 Python 和 Node SDK 都提供了流式接口，内部已经处理了连接管理和资源释放。

Python 示例：

from openai import OpenAI

client = OpenAI(api_key="sk-...", base_url="https://api.example.com/v1")

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "你好"}],
    stream=True,
)

try:
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)
except Exception as e:
    print("流中断:", e)
finally:
    stream.close()  # 显式关闭，确保底层连接释放

Node.js 示例：

import OpenAI from 'openai';

const client = new OpenAI({ apiKey: 'sk-...', baseURL: 'https://api.example.com/v1' });

const stream = await client.chat.completions.create({
  model: 'gpt-4o-mini',
  messages: [{ role: 'user', content: '你好' }],
  stream: true,
});

try {
  for await (const chunk of stream) {
    const delta = chunk.choices[0]?.delta?.content;
    if (delta) process.stdout.write(delta);
  }
} catch (err) {
  console.error('流中断:', err);
} finally {
  stream.controller.abort(); // 主动中止，释放连接
}

SDK 通常会自动处理 AbortController 和连接池，但仍建议在 finally 中显式调用关闭方法，以防提前退出循环。

使用原生 HTTP 客户端（需要手动管理）

如果不依赖 SDK，直接用 fetch（Node 18+）或 requests / httpx（Python）读取 SSE，就必须自己处理连接关闭。

Node.js 原生 fetch：

const controller = new AbortController();

const response = await fetch('https://api.example.com/v1/chat/completions', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    Authorization: 'Bearer sk-...',
  },
  body: JSON.stringify({
    model: 'gpt-4o-mini',
    messages: [{ role: 'user', content: '你好' }],
    stream: true,
  }),
  signal: controller.signal,
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

try {
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const text = decoder.decode(value, { stream: true });
    // 解析 SSE 行：以 data: 开头的 JSON
    for (const line of text.split('\n')) {
      if (line.startsWith('data: ')) {
        const data = line.slice(6);
        if (data === '[DONE]') {
          controller.abort(); // 提前结束，主动断开
          break;
        }
        const json = JSON.parse(data);
        const delta = json.choices[0]?.delta?.content;
        if (delta) process.stdout.write(delta);
      }
    }
  }
} catch (err) {
  if (err.name !== 'AbortError') console.error(err);
} finally {
  reader.cancel(); // 确保 reader 释放
}

Python httpx 流式读取：

import httpx
import json

with httpx.stream(
    "POST",
    "https://api.example.com/v1/chat/completions",
    headers={"Authorization": "Bearer sk-..."},
    json={
        "model": "gpt-4o-mini",
        "messages": [{"role": "user", "content": "你好"}],
        "stream": True,
    },
    timeout=30.0,
) as response:
    for line in response.iter_lines():
        if not line:
            continue
        if line.startswith("data: "):
            data = line[6:]
            if data == "[DONE]":
                break
            try:
                chunk = json.loads(data)
                delta = chunk["choices"][0]["delta"].get("content")
                if delta:
                    print(delta, end="", flush=True)
            except (json.JSONDecodeError, KeyError):
                pass
    # with 块退出时自动关闭连接，即使 break 提前退出也会触发

关键点：使用 with 语句或 try/finally 确保无论是否正常结束，连接都会被关闭。

流提前结束时的干净关闭

提前结束可能发生在：

• 用户点击“停止生成”
• 达到自定义的超时阈值
• 业务逻辑判断不需要继续接收
• 上游返回错误或 [DONE] 标记

Node.js 中的最佳实践：

1. 始终创建 AbortController，并传入 fetch 的 signal。
2. 在 finally 块中调用 controller.abort()，即使流已自然结束，abort 也是安全的（会静默忽略）。
3. 对于 ReadableStream，在 finally 中调用 reader.cancel()，显式释放锁。

Python 中的最佳实践：

1. 使用 with httpx.stream(...) as response: 上下文管理器，它保证连接关闭。
2. 如果手动管理连接，务必在 finally 中调用 response.close()。
3. 对于 requests，使用 stream=True 后同样需要 response.close()，否则连接会停留在连接池中。
4. OpenAI SDK 的 Stream 对象提供 .close() 方法，在 finally 中调用。

常见陷阱与检查清单

• 忘记处理异常路径：只在正常流程中关闭连接，异常时泄漏。一定要用 try/finally。
• 依赖垃圾回收：Python 的 del 或 Node 的 GC 不保证及时关闭底层 socket。
• 错误地重用响应对象：流式响应只能消费一次，重复读取会报错或挂起。
• 超时设置不当：连接超时和读取超时应分别设置，避免长时间等待导致资源占用。
• 未处理 [DONE] 标记：某些服务端会发送 data: [DONE] 后保持连接，客户端应主动断开。

检查清单：

• [ ] 每个流式请求都有对应的 AbortController 或 with 块。
• [ ] finally 中调用了 close() / abort() / cancel()。
• [ ] 捕获了 AbortError 并忽略它（这是预期行为）。
• [ ] 设置了合理的连接和读取超时。
• [ ] 在生产环境中监控打开的文件描述符数量。

总结

使用官方 SDK 可以大幅减少连接管理的工作量，但理解底层的 SSE 读取和关闭机制仍然重要。原生 HTTP 客户端给了你更多控制权，但也要求你显式管理资源。无论选择哪种方式，核心原则都是：确保每条流在离开作用域前都被关闭，无论是正常结束还是提前中断。

在聚合 API 的场景中，一个 API key 可能同时调用多个模型，流式请求更加频繁，连接泄漏的风险也更高。养成良好的关闭习惯，能让你的服务更稳定。