Socket.io应用AWS EC2带宽过高问题排查与优化咨询

问题根源确认 #

是的，你的高带宽消耗完全由重复且冗余的Socket.emit操作导致，核心问题包括：

• 后端每10秒向所有客户端广播20+个独立事件（如USDPrice/EURPrice），每个事件都携带1000条历史数据，这些数据仅货币字段不同，时间戳等核心信息重复传输
• 每个客户端（即使是只展示单一货币的页面）会接收所有20+个事件的完整数据，产生大量无效传输
• 后端在setInterval内重复绑定io.on("connection")，会导致新连接事件被多次监听，引发额外资源消耗

优化方案 #

后端优化：合并事件+按需推送 #

方案1：合并所有历史数据为单个事件

将所有货币的历史数据整合成一个结构发送，避免重复传输公共数据：

// 把connection监听移到定时任务外，仅绑定一次
io.on("connection", (sock) => {
  // 可选：后续可扩展客户端订阅逻辑
});

setInterval(async function () {
  const d = new Date();
  let allCurrencyInBlack = await getCurrencyInBlack();
  if (!allCurrencyInBlack) return;

  prices.push({
    time: Math.round(d.getTime() / 1000),
    value: allCurrencyInBlack,
  });

  // 预处理最近1000条有效数据
  const recentPrices = prices.slice(-1000).filter(item => item.value != null);
  
  // 单次发送包含所有货币的最新价格+历史数据
  io.sockets.emit("allCurrencyData", {
    latestPrices: allCurrencyInBlack,
    history: recentPrices.map(item => ({
      time: item.time,
      currencies: item.value
    }))
  });

  // 每日统计逻辑保持精简，合并为单个事件发送
  let usdCurrentBlackPrice = allCurrencyInBlack.USD;
  let dailyStats = {};
  if (d.getHours() === 21 && d.getMinutes() === 0) {
    openingPrice = usdCurrentBlackPrice;
    openingTime = Math.round(d.getTime() / 1000);
  }
  if (d.getHours() === 20 && d.getMinutes() === 59) {
    closingPrice = usdCurrentBlackPrice;
  }
  if (openingPrice != null) {
    dailyStats.dailyPercentage = ((usdCurrentBlackPrice - openingPrice) / openingPrice) * 100;
    const todaysPrices = recentPrices
      .filter(item => item.time > openingTime)
      .map(item => item.value.USD)
      .filter(v => v != null);
    if (todaysPrices.length > 0) {
      dailyStats.range = {
        max: Math.max(...todaysPrices).toFixed(2),
        min: Math.min(...todaysPrices).toFixed(2)
      };
    }
  }
  dailyStats.openingPrice = openingPrice;
  dailyStats.closingPrice = closingPrice;
  
  if (usdCurrentBlackPrice != null) {
    io.sockets.emit("dailyStats", dailyStats);
  }
}, 10000);

方案2：按需推送（极致带宽优化）

让客户端连接时声明所需货币，后端仅推送对应数据：

const subscriptions = new Map(); // 存储socket.id与订阅货币的映射

io.on("connection", (sock) => {
  // 接收客户端的订阅请求
  sock.on("subscribe", (currencies) => {
    subscriptions.set(sock.id, currencies);
    // 首次连接发送该货币的历史数据
    const recentPrices = prices.slice(-1000).filter(item => item.value != null);
    const filteredHistory = recentPrices.map(item => ({
      time: item.time,
      values: currencies.reduce((acc, curr) => {
        acc[curr] = item.value[curr];
        return acc;
      }, {})
    }));
    sock.emit("targetCurrencyHistory", filteredHistory);
  });

  // 断开连接时清除订阅
  sock.on("disconnect", () => {
    subscriptions.delete(sock.id);
  });
});

setInterval(async function () {
  const d = new Date();
  let allCurrencyInBlack = await getCurrencyInBlack();
  if (!allCurrencyInBlack) return;

  prices.push({
    time: Math.round(d.getTime() / 1000),
    value: allCurrencyInBlack,
  });

  // 仅向订阅的客户端推送所需数据
  subscriptions.forEach((currencies, socketId) => {
    const sock = io.sockets.sockets.get(socketId);
    if (!sock) return;
    // 推送最新价格
    sock.emit("latestPrices", currencies.reduce((acc, curr) => {
      acc[curr] = allCurrencyInBlack[curr];
      return acc;
    }, {}));
    // 仅推送最新一条数据（替代全量历史，进一步减少传输）
    sock.emit("newPrice", {
      time: Math.round(d.getTime() / 1000),
      values: currencies.reduce((acc, curr) => {
        acc[curr] = allCurrencyInBlack[curr];
        return acc;
      }, {})
    });
  });

  // 每日统计逻辑保持不变
}, 10000);

前端优化：只处理所需数据 #

适配合并事件的前端代码（以USD页面为例）

const sock = io.connect();
sock.on('allCurrencyData', data => {
  const { latestPrices, history } = data;
  // 更新实时价格
  document.querySelector('#livePrice').innerHTML = `${latestPrices.USD.toFixed(4)} Br`;
  const d = new Date();
  document.querySelector('#date').innerHTML = `${d.getDate()}/${d.getMonth() + 1}/${d.getFullYear()} - ${d.getHours()}:${d.getMinutes()}`;
  // 提取USD的历史数据
  const usdHistory = history.map(item => ({
    time: item.time,
    value: item.currencies.USD
  })).filter(item => item.value != null);
  lineSeries.setData(usdHistory.sort((a, b) => a.time - b.time));
});

// 监听每日统计数据
sock.on('dailyStats', stats => {
  // 按需更新开盘价、收盘价等元素
});

适配按需推送的前端代码

const sock = io.connect();
// 连接成功后订阅当前页面所需货币（如USD）
sock.emit("subscribe", ["USD"]);

// 接收历史数据初始化图表
sock.on('targetCurrencyHistory', history => {
  lineSeries.setData(history.sort((a, b) => a.time - b.time));
});

// 接收最新价格更新
sock.on('latestPrices', prices => {
  document.querySelector('#livePrice').innerHTML = `${prices.USD.toFixed(4)} Br`;
  const d = new Date();
  document.querySelector('#date').innerHTML = `${d.getDate()}/${d.getMonth() + 1}/${d.getFullYear()} - ${d.getHours()}:${d.getMinutes()}`;
});

// 接收单条新数据，仅更新图表而不需要重绘全量数据
sock.on('newPrice', price => {
  lineSeries.addData(price);
});

额外优化建议 #

• 减少历史数据量：将1000条数据缩减为300条，或按时间范围（如最近1小时）筛选
• 延长更新间隔：汇率无需每10秒更新，调整为30秒/1分钟一次可大幅降低传输频率
• 启用Socket.io压缩：初始化时开启自动压缩，减少数据体积

const io = require('socket.io')(server, {
  perMessageDeflate: {
    threshold: 1024 // 大于1KB的数据自动压缩
  }
});

内容的提问来源于stack exchange，提问作者gemchu geme