1. 使用标准库 socket 和 threading：最基础的方法，能让你完全理解 Telnet 服务器的底层工作原理,但需要自己处理很多细节。
2. 使用第三方库 pytelnetlib：比 socket 方法更高级一些，封装了部分 Telnet 协议的细节,但功能相对有限。
3. 使用第三方库 Telnetlib3：功能最强大、最健壮的库，推荐在生产环境中使用，它支持异步操作、并发连接、完整的 Telnet 协议（如选项协商）等。

使用标准库 socket 和 threading (最基础)

这种方法不依赖任何第三方库，能帮助你理解 Telnet 服务器是如何工作的，Telnet 本质上是一个基于文本的、明文的协议，服务器在指定端口监听客户端连接,接收和发送字符串数据。

核心概念

• socket：用于网络通信的底层接口。
• threading：为每个客户端连接创建一个新线程，这样服务器就可以同时处理多个客户端,而不会互相阻塞。

代码示例

这是一个简单的 Telnet 服务器，它会向每个连接的客户端显示欢迎信息，并回显客户端发送的任何内容，直到客户端输入 exit 或断开连接。

import socket
import threading
# --- 配置 ---
HOST = '0.0.0.0'  # 监听所有可用的网络接口
PORT = 23         # Telnet 默认端口
# --- 处理客户端连接的函数 ---
def handle_client(conn, addr):
    """处理单个客户端连接的函数"""
    print(f"[新连接] {addr} 已连接。")
    # 发送欢迎信息
    welcome_msg = "欢迎来到简单的 Python Telnet 服务器！\r\n"
    conn.sendall(welcome_msg.encode('utf-8'))
    try:
        while True:
            # 接收客户端数据 (recv 是阻塞的)
            data = conn.recv(1024)
            if not data:
                # 如果没有收到数据，说明客户端已断开连接
                break
            # 将接收到的字节转换为字符串
            message = data.decode('utf-8').strip()
            print(f"[来自 {addr}] {message}")
            # 检查退出命令
            if message.lower() == 'exit':
                response = "再见！\r\n"
                conn.sendall(response.encode('utf-8'))
                break
            # 回显消息
            response = f"你说了: '{message}'\r\n"
            conn.sendall(response.encode('utf-8'))
    except ConnectionResetError:
        print(f"[客户端断开] {addr} 强制断开了连接。")
    finally:
        # 确保连接被关闭
        conn.close()
        print(f"[连接关闭] {addr} 的连接已关闭。")
# --- 主服务器函数 ---
def start_server():
    """启动 Telnet 服务器"""
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 设置 SO_REUSEADDR 选项，允许地址在短时间内重用
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        server.bind((HOST, PORT))
        server.listen(5)  # 允许最多 5 个待处理连接
        print(f"服务器正在监听 {HOST}:{PORT}...")
        while True:
            # 接受新的客户端连接
            conn, addr = server.accept()
            # 为每个新连接创建一个新线程
            client_thread = threading.Thread(target=handle_client, args=(conn, addr))
            client_thread.start()
            print(f"[活动连接] {threading.active_count() - 1} 个客户端正在连接。")
    except KeyboardInterrupt:
        print("\n服务器正在关闭...")
    finally:
        server.close()
        print("服务器已关闭。")
if __name__ == "__main__":
    start_server()

如何运行和测试

1. 将代码保存为 simple_telnet_server.py。
2. 在终端中运行：python simple_telnet_server.py。
3. 你会看到 "服务器正在监听..." 的提示。
4. 打开另一个终端，使用 telnet 命令连接：telnet 127.0.0.1 23 (如果你的服务器在同一台机器上)。
5. 你将看到欢迎信息，输入任何文本并按回车，服务器会回显你的内容，输入 exit 并按回车,连接将断开。

使用第三方库 pytelnetlib (更简单)

pytelnetlib 是对 Python 标准库 telnetlib 的一个简单封装,使其更容易作为服务器使用。

安装

pip install pytelnetlib

代码示例

这个例子创建了一个回显服务器，并处理了 Telnet 的选项协商（比如是否启用回显）。

from pytelnetlib.server import TelnetServer
# 定义一个处理连接的类
class EchoServer:
    def __init__(self):
        self.users = {} # 可以用来存储连接状态
    # 当新客户端连接时调用
    def login(self, host, port):
        print(f"新连接来自 {host}:{port}")
        return True # 允许连接
    # 当客户端发送数据时调用
    def process_command(self, command, conn):
        # command 是一个字节串
        message = command.decode('utf-8').strip()
        print(f"收到命令: {message}")
        if message.lower() == 'exit':
            conn.send(b"再见！\r\n")
            return False # 返回 False 以断开连接
        # 默认行为是回显
        response = f"回显: {message}\r\n"
        conn.send(response.encode('utf-8'))
        return True # 保持连接
# --- 启动服务器 ---
if __name__ == '__main__':
    server = TelnetServer(port=2323, # 使用非默认端口以避免冲突
                          login_handler=EchoServer().login,
                          command_handler=EchoServer().process_command)
    print("pytelnetlib 服务器启动在端口 2323...")
    server.run()

使用第三方库 Telnetlib3 (推荐，功能最强)

Telnetlib3 是一个功能非常全面的 Telnet 库，支持异步 I/O (asyncio)，性能更好，并且支持完整的 Telnet 协议选项协商，这是构建健壮 Telnet 服务的首选。

安装

pip install telnetlib3

代码示例

这个例子展示了如何创建一个支持异步操作的服务器。

import asyncio
from telnetlib3.server import TelnetServer
# 这是一个异步的命令处理器
async def shell(reader, writer):
    """
    为每个连接创建一个异步的 shell 会话。
    reader: 用于读取客户端数据
    writer: 用于向客户端写入数据
    """
    host, port = writer.get_extra_info('peername')
    print(f"新连接来自 {host}:{port}")
    # 发送欢迎横幅
    writer.write("欢迎使用 Telnetlib3 异步服务器！\r\n")
    writer.write("输入 'help' 查看可用命令，'exit' 退出，\r\n")
    writer.write(">>> ")
    await writer.drain() # 确保数据被发送
    while not reader.at_eof():
        # 异步读取一行数据
        line = await reader.readline()
        if not line:
            break # 客户端断开连接
        command = line.decode('utf-8').strip()
        print(f"来自 {host}:{port} 的命令: {command}")
        if command == 'exit':
            writer.write("再见！\r\n")
            break
        elif command == 'help':
            response = "可用命令: help, exit, echo <text>\r\n"
        elif command.startswith('echo '):
            response = f"{command[5:]}\r\n"
        else:
            response = f"未知命令: {command}\r\n"
        writer.write(response)
        writer.write(">>> ")
        await writer.drain() # 再次刷新输出缓冲区
    print(f"连接 {host}:{port} 已关闭。")
    writer.close()
    await writer.wait_closed()
# --- 启动服务器 ---
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    # 创建一个 TelnetServer 实例，并指定我们的异步 shell 处理器
    coro = TelnetServer(port=2323, shell=shell)
    server = loop.run_until_complete(coro)
    try:
        print("Telnetlib3 异步服务器启动在端口 2323...")
        loop.run_forever() # 永久运行服务器
    except KeyboardInterrupt:
        pass
    finally:
        print("\n服务器正在关闭...")
        server.close()
        loop.run_until_complete(server.wait_closed())
        loop.close()

总结与对比

给你的建议：

• 如果你是初学者，想学习底层原理：从 方法一 开始,是最好的实践。
• 如果你只是想快速写一个简单的、功能单一的 Telnet 服务：可以考虑 方法二，它比 socket 更简洁。
• 如果你在开发一个需要稳定、高效、功能完善的生产级 Telnet 服务：强烈推荐方法三 (Telnetlib3),它能让你事半功倍。