Express路由中Socket.IO定向发消息及连接状态检测问题

问题1：仅向发起请求的用户发送消息 #

当前用io.emit()会给所有连接客户端广播消息，要实现定向推送，需关联HTTP请求与对应Socket连接：

客户端修改 #

发起请求时携带自身Socket ID：

import * as React from 'react';
import { io } from 'socket.io-client';

const socket = io('http://localhost:3001');

function DownloadComponent() {
  const startDownload = (magnetLink) => {
    fetch(`http://localhost:3001/test?magnetlink=${encodeURIComponent(magnetLink)}&socketId=${socket.id}`)
      .then(res => res.text())
      .catch(err => console.error(err));
  };

  React.useEffect(() => {
    socket.on('started', (data) => {
      console.log('下载开始:', data.name);
    });
    socket.on('data', (data) => {
      console.log(`下载进度: ${data.progress.toFixed(2)}%`);
    });
    socket.on('done', (data) => {
      console.log('下载完成:', data.name);
    });
    socket.on('error', (data) => {
      console.error('下载出错:', data.message);
    });

    return () => {
      // 组件卸载/页面刷新时发送取消请求
      socket.emit('cancel-torrent');
    };
  }, []);

  return (
    <button onClick={() => startDownload('你的磁力链接')}>开始下载</button>
  );
}

export default DownloadComponent;

服务端修改 #

维护Socket与Torrent的映射，定向发送消息：

const WebTorrent = require('webtorrent')
const client = new WebTorrent()
const express = require('express')
const app = express()
const port = 3001
const http = require('http')
const server = http.createServer(app)
const { Server } = require("socket.io");
const io = new Server(server, {
    cors: {
        origin: "http://localhost:3000"
    }
});

// 存储socketId到torrent实例的映射
const socketTorrentMap = new Map();

// 监听socket连接与断开事件
io.on('connection', (socket) => {
  // 处理主动取消下载请求
  socket.on('cancel-torrent', () => {
    const torrent = socketTorrentMap.get(socket.id);
    if (torrent) {
      torrent.destroy(() => {
        console.log(`已终止socket ${socket.id}的下载任务`);
        socketTorrentMap.delete(socket.id);
      });
    }
  });

  // 客户端断开时自动终止对应下载
  socket.on('disconnect', () => {
    const torrent = socketTorrentMap.get(socket.id);
    if (torrent) {
      torrent.destroy(() => {
        console.log(`socket ${socket.id}断开，已清理对应下载任务`);
        socketTorrentMap.delete(socket.id);
      });
    }
  });
});

app.get('/test', (req, res) => {
  const { magnetlink, socketId } = req.query;
  if (!magnetlink || !socketId) {
    return res.status(400).send('缺少magnetlink或socketId参数');
  }

  // 校验socket是否有效
  const targetSocket = io.sockets.sockets.get(socketId);
  if (!targetSocket) {
    return res.status(400).send('无效的socket连接');
  }

  // 避免同一socket重复创建任务
  if (socketTorrentMap.has(socketId)) {
    return res.status(400).send('当前已有正在进行的下载任务');
  }

  client.add(magnetlink, (torrent) => {
    socketTorrentMap.set(socketId, torrent);
    const torrentName = torrent.name;

    // 仅向发起请求的socket发送消息
    targetSocket.emit('started', { name: torrentName });

    // 修正原代码语法错误，补充进度数据
    torrent.on('download', (bytes) => {
      const progress = (torrent.downloaded / torrent.length) * 100;
      targetSocket.emit('data', {
        name: torrentName,
        progress,
        downloaded: torrent.downloaded,
        total: torrent.length
      });
    });

    torrent.on('done', () => {
      targetSocket.emit('done', { name: torrentName });
      socketTorrentMap.delete(socketId);
    });

    // 处理下载错误
    torrent.on('error', (err) => {
      targetSocket.emit('error', { name: torrentName, message: err.message });
      socketTorrentMap.delete(socketId);
    });
  });

  res.send('下载任务已启动');
})

server.listen(port, () => {
  console.log(`Running on http://localhost:${port}`)
})

问题2：检测用户连接并终止闲置下载 #

上述代码已通过以下逻辑解决：

• 用socketTorrentMap绑定每个socket与对应的下载任务
• 监听socket的disconnect事件，断开时自动销毁对应torrent并清理映射
• 支持客户端主动发送cancel-torrent事件取消下载
• 下载完成/出错时自动清理映射，避免内存泄漏

额外修正说明 #

• 修复了原代码中torrent.on('download, () => {的引号缺失语法错误
• 修正了emit('data', () => { name: name })错误传递函数的问题，改为直接传递进度对象
• 新增下载错误处理，避免任务异常时内存泄漏

内容的提问来源于stack exchange，提问作者Wamy-Dev