Django + xterm.js 实现多客户端独立终端会话的完整解决方案-Python教程-PHP中文网

Django + xterm.js 实现多客户端独立终端会话的完整解决方案

本文详解如何在 django 后端与 xterm.js 前端组合中，为每个浏览器客户端分配独立的伪终端（pty）会话，避免多用户共享同一终端导致的输入/输出混乱问题。核心在于会话隔离、资源生命周期管理及 socket.io 的正确事件绑定。

本文详解如何在 django 后端与 xterm.js 前端组合中，为每个浏览器客户端分配独立的伪终端（pty）会话，避免多用户共享同一终端导致的输入/输出混乱问题。核心在于会话隔离、资源生命周期管理及 socket.io 的正确事件绑定。

当前代码最大的根本性问题是全局共享 PTY 句柄（fd 和 child_pid），所有客户端连接都复用同一套终端进程，导致输入混杂、输出错乱、状态冲突。解决思路不是“修补”单例逻辑，而是彻底转向会话级隔离模型：每个客户端连接对应一个独立的 PtyTerminal 实例，并通过唯一会话标识（如 sid）进行映射与管理。

✅ 步骤一：封装健壮的 PtyTerminal 类（推荐使用答案中提供的实现）

该类将 PTY 创建、读写、尺寸调整、生命周期控制等逻辑封装为可复用组件，显著提升可维护性与并发安全性：

# pty_terminal.py
import os
from select import select
from pty import fork
from termios import TIOCSWINSZ, TIOCGWINSZ
from struct import pack, unpack
from fcntl import ioctl
from os import execl, closerange
from time import sleep

DEFAULT_SIZE = (24, 80, 0, 0)
DEFAULT_CMD = ['/bin/bash']  # 替换为你实际需要的 shell 或命令

class PtyTerminal:
    def __init__(self, cmd=None, size=None):
        self.master = None
        self.child = None
        child, master = fork()
        if child == 0:
            closerange(3, 255)
            # 注意：subprocess.run 在子进程中会阻塞，应直接 execl
            execl(*(cmd or DEFAULT_CMD))
        self.child = child
        self.master = master
        self._size = (0, 0, 0, 0)
        self.size = size or DEFAULT_SIZE

    @property
    def size(self):
        return self._size

    @size.setter
    def size(self, value):
        if self.master and value != self._size:
            ioctl(self.master, TIOCSWINSZ, pack('HHHH', *value))
            try:
                self._size = unpack('HHHH', ioctl(self.master, TIOCGWINSZ, b'12345678'))
            except OSError:
                pass

    def read(self, length=4096, timeout=0.01):
        if not self.master:
            return b''
        try:
            rd, _, _ = select([self.master], [], [], timeout)
            return os.read(self.master, length) if rd else b''
        except OSError:
            self.wait_child()
            return b''

    def write(self, data):
        if not self.master or not data:
            return
        if isinstance(data, str):
            data = data.encode('utf-8')
        os.write(self.master, data)

    def wait_child(self):
        if self.child:
            try:
                os.waitpid(self.child, os.WNOHANG)
                self.child = None
            except ChildProcessError:
                self.child = None

    def close(self):
        if self.master:
            os.close(self.master)
            self.master = None
        self.wait_child()

⚠️ 关键改进点：

使用 execl() 替代 subprocess.run()，避免子进程被父进程阻塞；

read() 方法支持超时，防止阻塞主线程；

封装 close() 和 wait_child()，确保资源及时释放；

移除全局变量依赖，实例化即隔离。

✅ 步骤二：基于 Socket.IO 的会话级终端管理（修改 views.py）

摒弃全局 fd/child_pid，改用字典 terminals: Dict[str, PtyTerminal] 按 sid 索引：

Feishu calendar sync, local ics to json data for AI agent

将ICS日历文件转为JSON格式，用于飞书日历导入导出及数据集成。

下载

# views.py
import eventlet
from django.shortcuts import render
import socketio
from .pty_terminal import PtyTerminal

async_mode = "eventlet"
sio = socketio.Server(async_mode=async_mode)

# 存储每个 sid 对应的终端实例
terminals = {}

def index(request):
    return render(request, "index.html")

@sio.event
def connect(sid, environ):
    # 为新连接创建专属终端
    terminal = PtyTerminal(cmd=['/bin/bash'])
    terminals[sid] = terminal
    print(f"[+] New terminal for sid={sid}")

    # 启动后台任务：持续读取并广播输出
    sio.start_background_task(
        target=read_and_forward_output,
        sid=sid,
        terminal=terminal
    )

@sio.event
def disconnect(sid):
    # 安全关闭对应终端
    if sid in terminals:
        terminals[sid].close()
        del terminals[sid]
        print(f"[-] Terminal closed for sid={sid}")

@sio.event
def resize(sid, message):
    if sid in terminals:
        terminals[sid].size = (message["rows"], message["cols"], 0, 0)

@sio.event
def pty_input(sid, message):
    if sid in terminals:
        terminals[sid].write(message["input"])

def read_and_forward_output(sid, terminal):
    """后台任务：向指定 sid 推送终端输出"""
    while sid in terminals and terminals[sid].master:
        try:
            output = terminal.read(timeout=0.02)
            if output:
                sio.emit("pty_output", {"output": output.decode('utf-8')}, room=sid)
        except Exception as e:
            print(f"Read error for {sid}: {e}")
            break
    # 连接断开或终端关闭后清理
    if sid in terminals:
        terminals[sid].close()
        del terminals[sid]

✅ 关键设计说明：

room=sid 确保 pty_output 仅推送给对应客户端，杜绝跨会话污染；

read_and_forward_output 作为协程任务运行，每个终端独占读取流；

disconnect 和异常退出均触发 close()，防止僵尸进程和文件描述符泄漏；

所有操作基于 sid 隔离，天然支持任意数量并发客户端。

✅ 步骤三：前端适配（index.html 关键修改）

确保前端建立连接后，只监听自身 sid 的事件（Socket.IO 默认即按 room 隔离，无需额外改动），但需注意：

移除全局 sessionStorage.clear()：它会误杀其他标签页的会话；
确保每次连接都触发 connect 事件（当前代码已满足）；
发送输入时无需修改，pty_input 事件自动携带当前 sid 上下文。

<!-- index.html 中保持原 socket.emit 逻辑即可 -->
term.onKey(e => {
    socket.emit("pty_input", { "input": e.key });
});

socket.on("pty_output", function (data) {
    term.write(data.output); // 输出已限定为本 sid
});

? 重要注意事项与最佳实践

安全加固：生产环境必须添加身份认证（如 JWT Token 校验）、命令白名单、超时自动销毁（如 10 分钟无活动则 close()）、最大并发数限制；
资源监控：定期检查 terminals 字典大小，对异常长时间存活的 sid 主动清理；
信号处理：考虑捕获 SIGCHLD 防止僵尸进程（Django 进程中需 signal.signal(signal.SIGCHLD, signal.SIG_DFL)）；
替代方案评估：若并发量极高（>1000），建议迁移到 asyncio + uvicorn + socketio.AsyncServer，性能更优；
调试技巧：在 PtyTerminal.__init__() 中添加 print(f"Spawned bash PID {self.child}")，配合 ps aux | grep bash 验证进程隔离性。

通过以上重构，你的终端应用将真正实现「一人一终端」——无论新标签页、新设备还是多用户同时访问，彼此完全独立、互不干扰，兼具健壮性与可扩展性。