在生产环境中使用 WebSocket 连接大模型 API 时，连接泄漏是最容易被忽视但破坏力最强的隐患之一。一次未关闭的连接可能在数小时内耗尽服务器文件描述符，导致整个服务瘫痪。我曾在某项目中遇到凌晨三点服务全部挂掉的紧急故障，排查后发现是 200+ 个 WebSocket 连接从未被正确关闭。本文将系统性地讲解如何检测和排查这类问题。

HolySheep vs 官方 API vs 其他中转站：核心差异对比

对比维度 HolySheep AI 官方 API 其他中转站
汇率优势 ¥1=$1，无损兑换 ¥7.3=$1（溢价 85%+） ¥5-8=$1（不稳定）
国内延迟 <50ms 直连 200-500ms（跨境） 80-300ms（不稳定）
充值方式 微信/支付宝 国际信用卡 参差不齐
免费额度 注册即送 $5 试用 极少或无
WebSocket 支持 完整流式输出 需要代理 部分支持
连接稳定性 企业级 SLA 依赖代理质量 良莠不齐

如果你需要稳定、低延迟、成本可控的 WebSocket AI 对接方案，立即注册 HolySheep AI，新用户首月赠送免费额度。

为什么 WebSocket 连接泄漏如此危险

我第一次深刻体会到连接泄漏的破坏力，是在部署一个实时对话系统时。那天晚上服务突然全部不可用，SSH 登录后发现错误信息：too many open files。排查日志发现 WebSocket 连接数在 6 小时内从正常值 50 个飙升到 12000+ 个。

连接泄漏的危害主要体现在三个层面：

基础 WebSocket 连接代码：规范写法

首先给出规范的 WebSocket 连接代码模板，这是我修复过数十个项目后总结的最佳实践：

const WebSocket = require('ws');

class HolySheepWebSocket {
  constructor(apiKey, onMessage, onError) {
    this.apiKey = apiKey;
    this.onMessage = onMessage;
    this.onError = onError;
    this.ws = null;
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 5;
    this.heartbeatInterval = null;
  }

  connect() {
    const url = 'wss://api.holysheep.ai/v1/chat/completions';
    const headers = {
      'Authorization': Bearer ${this.apiKey},
      'Content-Type': 'application/json'
    };

    this.ws = new WebSocket(url, { headers });

    // ⭐ 关键：设置超时保护
    const connectTimeout = setTimeout(() => {
      if (this.ws.readyState !== WebSocket.OPEN) {
        console.error('[连接超时] 10秒内未建立连接');
        this.ws.terminate();
      }
    }, 10000);

    this.ws.on('open', () => {
      clearTimeout(connectTimeout);
      console.log('[WebSocket] 连接已建立');
      this.startHeartbeat();
      this.reconnectAttempts = 0;
    });

    this.ws.on('message', (data) => {
      try {
        const message = JSON.parse(data);
        this.onMessage(message);
      } catch (e) {
        console.error('[解析错误]', e.message);
      }
    });

    this.ws.on('error', (error) => {
      console.error('[WebSocket 错误]', error.message);
      this.onError && this.onError(error);
    });

    this.ws.on('close', (code, reason) => {
      console.log([WebSocket 关闭] code=${code}, reason=${reason});
      this.cleanup();
    });

    this.ws.on('ping', () => {
      console.log('[心跳] 收到服务器 ping');
    });

    this.ws.on('pong', () => {
      console.log('[心跳] 收到服务器 pong');
    });
  }

  startHeartbeat() {
    // ⭐ 关键：每 30 秒发送心跳，防止连接被中间设备关闭
    this.heartbeatInterval = setInterval(() => {
      if (this.ws && this.ws.readyState === WebSocket.OPEN) {
        this.ws.ping();
        console.log('[心跳] 已发送 ping');
      }
    }, 30000);
  }

  cleanup() {
    // ⭐ 核心：确保清理所有资源
    if (this.heartbeatInterval) {
      clearInterval(this.heartbeatInterval);
      this.heartbeatInterval = null;
    }
    if (this.ws) {
      this.ws.removeAllListeners();
      if (this.ws.readyState === WebSocket.OPEN || this.ws.readyState === WebSocket.CONNECTING) {
        this.ws.close(1000, '正常关闭');
      }
      this.ws = null;
    }
  }

  send(message) {
    if (this.ws && this.ws.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(message));
    } else {
      console.error('[发送失败] 连接未就绪，当前状态:', this.ws?.readyState);
    }
  }

  destroy() {
    // ⭐ 关键：外部调用销毁时彻底清理
    console.log('[销毁连接] 正在清理所有资源...');
    this.cleanup();
  }
}

// 使用示例
const ws = new HolySheepWebSocket(
  'YOUR_HOLYSHEEP_API_KEY',  // 替换为你的 HolySheep API Key
  (msg) => console.log('[收到消息]', JSON.stringify(msg)),
  (err) => console.error('[错误回调]', err)
);

ws.connect();

// 确保进程退出时关闭所有连接
process.on('SIGTERM', () => {
  console.log('[进程退出] 正在关闭所有连接');
  ws.destroy();
  process.exit(0);
});

连接泄漏的五大常见原因

1. 异常分支未清理

这是最常见的泄漏原因。在 try-catch 块的 catch 分支中，如果没有显式关闭连接，异常时的连接就会泄漏。

2. 缺少心跳检测

WebSocket 连接如果长时间无数据传输，中间设备（如 NAT、负载均衡器）会主动断开空闲连接。业务代码以为连接还在，实际上已经是死连接。

3. 重连逻辑未限制

网络抖动时，快速重连可能导致指数级连接增长。我在 HolySheep AI 的生产环境监控中发现，某些客户在网络不稳定时会瞬间创建 500+ 并发连接。

4. 事件监听器未移除

每次重新连接时，如果旧的监听器没有清理，新消息会同时触发多个处理函数，内存占用翻倍增长。

5. 进程退出时未清理

SIGTERM 信号到来时，如果优雅关闭逻辑缺失，连接会直接被操作系统强制关闭，既不优雅也无法释放资源。

连接泄漏检测工具：完整监控代码

const WebSocket = require('ws');
const { EventEmitter } = require('events');

class ConnectionMonitor extends EventEmitter {
  constructor() {
    super();
    this.connections = new Map(); // connectionId -> connectionInfo
    this.metrics = {
      totalCreated: 0,
      totalClosed: 0,
      totalError: 0,
      activeCount: 0
    };
    this.startTime = Date.now();
    
    // 定期报告状态
    setInterval(() => this.report(), 60000);
  }

  createConnection(connectionId, options = {}) {
    const apiKey = options.apiKey || process.env.HOLYSHEEP_API_KEY;
    const baseUrl = 'wss://api.holysheep.ai/v1/chat/completions';
    
    const ws = new WebSocket(baseUrl, {
      headers: {
        'Authorization': Bearer ${apiKey},
        'X-Connection-Id': connectionId
      }
    });

    const connectionInfo = {
      id: connectionId,
      ws,
      createdAt: Date.now(),
      lastActivity: Date.now(),
      messageCount: 0,
      state: 'connecting',
      tags: options.tags || {}
    };

    this.connections.set(connectionId, connectionInfo);
    this.metrics.totalCreated++;
    this.metrics.activeCount++;
    
    console.log([Monitor] 创建连接 ${connectionId}，当前活跃: ${this.metrics.activeCount});
    this.emit('connection:created', connectionInfo);

    ws.on('open', () => {
      connectionInfo.state = 'open';
      console.log([Monitor] 连接 ${connectionId} 已打开);
    });

    ws.on('message', (data) => {
      connectionInfo.lastActivity = Date.now();
      connectionInfo.messageCount++;
    });

    ws.on('close', (code, reason) => {
      connectionInfo.state = 'closed';
      connectionInfo.closeCode = code;
      connectionInfo.closeReason = reason?.toString() || '';
      this.removeConnection(connectionId);
      console.log([Monitor] 连接 ${connectionId} 已关闭(code=${code}));
    });

    ws.on('error', (error) => {
      connectionInfo.state = 'error';
      connectionInfo.lastError = error.message;
      this.metrics.totalError++;
      console.error([Monitor] 连接 ${connectionId} 错误: ${error.message});
      this.emit('connection:error', { connectionId, error });
    });

    return ws;
  }

  removeConnection(connectionId) {
    const info = this.connections.get(connectionId);
    if (info) {
      const lifetime = Date.now() - info.createdAt;
      console.log([Monitor] 移除连接 ${connectionId}，存活时间: ${lifetime}ms，消息数: ${info.messageCount});
      this.connections.delete(connectionId);
      this.metrics.totalClosed++;
      this.metrics.activeCount--;
      this.emit('connection:removed', { ...info, lifetime });
    }
  }

  // ⭐ 核心功能：检测可能的泄漏
  detectLeaks() {
    const now = Date.now();
    const potentialLeaks = [];
    const staleThreshold = 5 * 60 * 1000; // 5分钟无活动视为可疑
    
    for (const [id, info] of this.connections) {
      if (info.state === 'open') {
        const inactiveTime = now - info.lastActivity;
        
        if (inactiveTime > staleThreshold) {
          potentialLeaks.push({
            connectionId: id,
            inactiveTime,
            messageCount: info.messageCount,
            lifetime: now - info.createdAt,
            reason: inactiveTime > 10 * 60 * 1000 ? 'STALE_CONNECTION' : 'LONG_INACTIVE'
          });
        }
      }
    }

    if (potentialLeaks.length > 0) {
      console.warn([Monitor] ⚠️ 检测到 ${potentialLeaks.length} 个可疑连接:);
      potentialLeaks.forEach(leak => {
        console.warn(  - ${leak.connectionId}: 无活动 ${Math.round(leak.inactiveTime/1000)}秒，消息数 ${leak.messageCount});
      });
    }

    return potentialLeaks;
  }

  report() {
    const uptime = Date.now() - this.startTime;
    const leakCheck = this.detectLeaks();
    
    const report = {
      timestamp: new Date().toISOString(),
      uptime: ${Math.round(uptime / 1000)}秒,
      metrics: { ...this.metrics },
      activeConnections: this.connections.size,
      suspiciousLeaks: leakCheck.length,
      avgLifetime: this.calculateAvgLifetime(),
      messageRate: this.calculateMessageRate()
    };

    console.log('[Monitor] === 状态报告 ===');
    console.log(JSON.stringify(report, null, 2));
    
    // ⭐ 关键：活跃连接数异常时告警
    if (this.metrics.activeCount > 500) {
      console.error([Monitor] 🚨 严重告警：活跃连接数 ${this.metrics.activeCount} 超过阈值);
      this.emit('alert:high', report);
    }

    return report;
  }

  calculateAvgLifetime() {
    let total = 0;
    let count = 0;
    for (const info of this.connections.values()) {
      total += Date.now() - info.createdAt;
      count++;
    }
    return count > 0 ? Math.round(total / count) : 0;
  }

  calculateMessageRate() {
    const uptime = (Date.now() - this.startTime) / 1000;
    return uptime > 0 ? (this.metrics.totalCreated / uptime).toFixed(2) : 0;
  }

  // 关闭所有连接（清理用）
  closeAll() {
    console.log([Monitor] 正在关闭所有 ${this.connections.size} 个连接...);
    for (const [id, info] of this.connections) {
      if (info.ws && info.ws.readyState === WebSocket.OPEN) {
        info.ws.close(1000, 'monitor_shutdown');
      }
    }
  }
}

// 使用示例
const monitor = new ConnectionMonitor();

// 每 5 分钟执行泄漏检测
setInterval(() => {
  const leaks = monitor.detectLeaks();
  if (leaks.length > 0) {
    // 这里可以接入钉钉/飞书告警
    console.error('[告警] 发现连接泄漏，请检查！');
  }
}, 5 * 60 * 1000);

// 进程退出时清理
process.on('SIGTERM', () => {
  monitor.closeAll();
  process.exit(0);
});

module.exports = { ConnectionMonitor };

常见报错排查

错误 1：WebSocket connection closed with code 1006

错误信息：WebSocket connection closed: code=1006 (abnormal closure), reason=``

可能原因：

排查步骤：

// 添加详细日志诊断
ws.on('close', (code, reason) => {
  console.error('[连接关闭]', {
    code,
    reason: reason.toString(),
    wasClean: ws._closeFrameReceived,
    bufferedAmount: ws.bufferedAmount,
    readyState: ws.readyState
  });
  
  // 诊断常见 1006 原因
  if (code === 1006) {
    console.error('[诊断] 1006 通常表示：');
    console.error('  1. 认证失败 - 检查 API Key 是否正确');
    console.error('  2. 服务器拒绝 - 检查 baseUrl 是否正确');
    console.error('  3. 网络超时 - 增加重连间隔');
    console.error('  4. 协议错误 - 检查请求格式');
    
    // 验证连接（示例）
    verifyConnection().then(valid => {
      if (!valid) {
        console.error('[致命] 连接验证失败，请检查配置');
        process.exit(1);
      }
    });
  }
});

async function verifyConnection() {
  const https = require('https');
  const apiKey = 'YOUR_HOLYSHEEP_API_KEY'; // 替换为你的 Key
  
  // 验证 API Key 有效性
  return new Promise((resolve) => {
    const options = {
      hostname: 'api.holysheep.ai',
      port: 443,
      path: '/v1/models',
      method: 'GET',
      headers: {
        'Authorization': Bearer ${apiKey}
      },
      timeout: 5000
    };

    const req = https.request(options, (res) => {
      resolve(res.statusCode === 200);
    });

    req.on('error', (e) => {
      console.error('[验证失败]', e.message);
      resolve(false);
    });

    req.on('timeout', () => {
      req.destroy();
      resolve(false);
    });

    req.end();
  });
}

错误 2：Socket hang up / ECONNRESET

错误信息：Error: socket hang up 或 Error: read ECONNRESET

可能原因：

解决方案：实现连接池和指数退避重试

class ResilientConnectionPool {
  constructor(options = {}) {
    this.maxConnections = options.maxConnections || 10;
    this.baseDelay = options.baseDelay || 1000;
    this.maxDelay = options.maxDelay || 30000;
    this.connections = [];
    this.pendingRequests = [];
    this.stats = {
      totalRequests: 0,
      successfulRequests: 0,
      failedRequests: 0,
      totalRetries: 0
    };
  }

  async acquire() {
    this.stats.totalRequests++;
    
    // 找到空闲连接
    const idleConnection = this.connections.find(c => c.busy === false);
    
    if (idleConnection) {
      idleConnection.busy = true;
      return idleConnection;
    }
    
    // 如果没达到上限，创建新连接
    if (this.connections.length < this.maxConnections) {
      const conn = await this.createConnection();
      conn.busy = true;
      this.connections.push(conn);
      return conn;
    }
    
    // 等待空闲连接
    return new Promise((resolve) => {
      this.pendingRequests.push(resolve);
    });
  }

  async createConnection() {
    const ws = new WebSocket('wss://api.holysheep.ai/v1/chat/completions', {
      headers: {
        'Authorization': Bearer ${process.env.HOLYSHEEP_API_KEY}
      }
    });

    return new Promise((resolve, reject) => {
      const timeout = setTimeout(() => {
        ws.terminate();
        reject(new Error('连接超时'));
      }, 10000);

      ws.on('open', () => {
        clearTimeout(timeout);
        resolve({
          ws,
          busy: false,
          createdAt: Date.now(),
          useCount: 0
        });
      });

      ws.on('error', (err) => {
        clearTimeout(timeout);
        reject(err);
      });
    });
  }

  release(connection) {
    connection.busy = false;
    connection.useCount++;
    
    // 检查连接是否需要重建
    if (connection.useCount > 100 || Date.now() - connection.createdAt > 30 * 60 * 1000) {
      console.log('[连接池] 重建老化连接');
      this.connections = this.connections.filter(c => c !== connection);
      connection.ws.close();
      return;
    }

    // 如果有待处理请求，分配给它
    if (this.pendingRequests.length > 0) {
      const pendingResolve = this.pendingRequests.shift();
      connection.busy = true;
      pendingResolve(connection);
    }
  }

  async executeWithRetry(requestFn, maxRetries = 3) {
    let lastError;
    
    for (let attempt = 0; attempt < maxRetries; attempt++) {
      let connection;
      
      try {
        connection = await this.acquire();
        
        // ⭐ 关键：使用连接执行请求
        const result = await requestFn(connection.ws);
        
        this.stats.successfulRequests++;
        this.release(connection);
        return result;
        
      } catch (error) {
        this.stats.failedRequests++;
        
        if (connection) {
          // 出错了，标记连接需要重建
          this.connections = this.connections.filter(c => c !== connection);
          connection.ws.terminate();
        }

        // 指数退避
        if (attempt < maxRetries - 1) {
          const delay = Math.min(
            this.baseDelay * Math.pow(2, attempt),
            this.maxDelay
          );
          this.stats.totalRetries++;
          console.log([重试] 第 ${attempt + 1} 次失败，${delay}ms 后重试: ${error.message});
          await new Promise(r => setTimeout(r, delay));
        } else {
          lastError = error;
        }
      }
    }
    
    throw lastError;
  }

  // 关闭所有连接
  closeAll() {
    console.log('[连接池] 关闭所有连接');
    for (const conn of this.connections) {
      conn.ws.close(1000, 'pool_shutdown');
    }
    this.connections = [];
    this.pendingRequests = [];
  }
}

// 使用示例
const pool = new ResilientConnectionPool({ maxConnections: 5 });

async function sendMessage(message) {
  return pool.executeWithRetry(async (ws) => {
    return new Promise((resolve, reject) => {
      let response = '';
      
      ws.on('message', (data) => {
        response += data.toString();
      });
      
      ws.on('error', reject);
      ws.on('close', () => resolve(response));
      
      ws.send(JSON.stringify({
        model: 'gpt-4',
        messages: [{ role: 'user', content: message }],
        stream: true
      }));
      
      // 超时保护
      setTimeout(() => reject(new Error('请求超时')), 60000);
    });
  });
}

错误 3：MaxListenersExceededWarning

警告信息：MaxListenersExceededWarning: Possible EventEmitter memory leak detected

可能原因：重复添加事件监听器但未移除，导致监听器数量线性增长。

解决方案：

// ❌ 错误写法：每次调用都会添加新的监听器
function onMessage(ws, message) {
  ws.on('data', (data) => {
    console.log(data);
  });
  // 如果这个函数被调用多次，监听器会不断累积
}

// ✅ 正确写法：使用 once 或显式管理
class WebSocketHandler {
  constructor(ws) {
    this.ws = ws;
    this.setupListeners();
  }

  setupListeners() {
    // 方案 1：使用 once 而不是 on
    this.ws.once('message', this.handleMessage.bind(this));
    this.ws.once('close', this.handleClose.bind(this));
    this.ws.once('error', this.handleError.bind(this));
    
    // 方案 2：如果必须用 on，先 remove 再 add
    // this.ws.removeAllListeners('message');
    // this.ws.on('message', this.handleMessage.bind(this));
    
    // 方案 3：设置合理的监听器上限
    this.ws.setMaxListeners(10);
  }

  handleMessage(data) {
    console.log('[收到]', data.toString());
  }

  handleClose() {
    console.log('[连接关闭]');
  }

  handleError(err) {
    console.error('[错误]', err.message);
  }
}

// ✅ 全局内存泄漏检测（生产环境必须开启）
process.on('warning', (warning) => {
  if (warning.name === 'MaxListenersExceededWarning') {
    console.error('🚨 检测到监听器泄漏！');
    console.error(warning.message);
    
    // 记录堆栈追踪
    console.error('堆栈:', warning.stack);
    
    // 可以选择强制退出以避免内存持续增长
    // process.exit(1);
  }
});

// 检查监听器数量
setInterval(() => {
  const listenerCount = (emitter) => {
    return emitter.eventNames().reduce((count, event) => {
      return count + emitter.listenerCount(event);
    }, 0);
  };

  // 监控所有 WebSocket 实例
  if (global.wsInstances) {
    for (const [id, ws] of global.wsInstances) {
      const count = listenerCount(ws);
      if (count > 20) {
        console.warn([警告] WebSocket ${id} 有 ${count} 个监听器，可能存在泄漏);
      }
    }
  }
}, 60000);

资源泄漏的完整排查流程

根据我的实战经验，排查资源泄漏需要系统性的方法。以下是我总结的排查流程：

  1. 确认泄漏现象：监控连接数是否持续增长
  2. 定位泄漏点：通过堆栈追踪找到未关闭的连接
  3. 分析泄漏模式：是偶发还是必然，是否与特定操作相关
  4. 修复并验证：修复后压测确认问题解决
  5. 添加监控：防止未来再次出现
// 完整的泄漏检测脚本
const fs = require('fs');
const os = require('os');

// 1. 基础监控：文件描述符
function getOpenFDs() {
  const pid = process.pid;
  try {
    const fds = fs.readdirSync(/proc/${pid}/fd);
    return fds.length;
  } catch {
    return -1;
  }
}

// 2. 内存监控
function getMemoryUsage() {
  const used = process.memoryUsage();
  return {
    heapUsed: Math.round(used.heapUsed / 1024 / 1024) + 'MB',
    heapTotal: Math.round(used.heapTotal / 1024 / 1024) + 'MB',
    rss: Math.round(used.rss / 1024 / 1024) + 'MB',
    external: Math.round(used.external / 1024 / 1024) + 'MB'
  };
}

// 3. 事件循环监控
function getEventLoopLag() {
  return new Promise((resolve) => {
    const start = process.hrtime.bigint();
    setImmediate(() => {
      const lag = Number(process.hrtime.bigint() - start) / 1e6;
      resolve(Math.round(lag));
    });
  });
}

// 4. 完整健康检查
async function healthCheck() {
  console.log('\n=== 健康检查报告 ===');
  console.log('时间:', new Date().toISOString());
  
  const openFDs = getOpenFDs();
  console.log(文件描述符: ${openFDs} (上限: 1024));
  
  if (openFDs > 800) {
    console.error('🚨 警告: 文件描述符使用率超过 80%');
  }

  const mem = getMemoryUsage();
  console.log('内存使用:', JSON.stringify(mem));
  
  const lag = await getEventLoopLag();
  console.log(事件循环延迟: ${lag}ms);
  if (lag > 100) {
    console.error('⚠️ 警告: 事件循环延迟过高，可能存在阻塞');
  }

  // 检查 WebSocket 统计
  if (global.connectionMonitor) {
    const monitor = global.connectionMonitor;
    console.log('WebSocket 活跃连接:', monitor.metrics.activeCount);
    console.log('WebSocket 总创建:', monitor.metrics.totalCreated);
    console.log('WebSocket 总关闭:', monitor.metrics.totalClosed);
    
    if (monitor.metrics.totalCreated - monitor.metrics.totalClosed > 100) {
      console.error('🚨 严重: 存在未关闭的 WebSocket 连接！');
    }
  }

  console.log('====================\n');
}

// 每 30 秒执行一次健康检查
setInterval(healthCheck, 30000);

// 导出供外部调用
module.exports = { healthCheck, getOpenFDs, getMemoryUsage, getEventLoopLag };

HolySheep AI 的 WebSocket 优化建议

在使用 HolySheep AI 的 WebSocket API 时，以下是几点特别建议：

目前 HolySheep AI 的主流模型价格如下（2026 年最新）：

相比官方 API 7.3 倍的汇率差，使用 HolySheep AI 可以节省超过 85% 的成本。

总结

WebSocket 连接泄漏是生产环境中常见但容易被忽视的问题。通过本文介绍的方法，你可以：

建议将本文提供的监控代码集成到你的生产环境中，并设置合理的告警阈值。我个人习惯设置两级告警：活跃连接超过 200 时发 warning，超过 500 时发 critical 并自动触发连接清理。

如果你还没有尝试过 HolySheep AI，强烈建议体验一下。国内直连 <50ms 的延迟和 ¥1=$1 的汇率优势，配合稳定的 WebSocket 支持，可以让你的 AI 应用性能和成本都得到优化。

👉 免费注册 HolySheep AI，获取首月赠额度

相关资源

相关文章