面试题：Python 中如何使用 socket 模块创建一个基本的高性能 TCP 服务器框架

TCP 服务器示例代码

import socket

# 创建 socket 对象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 获取本地主机名
host = socket.gethostname()
port = 9999

# 绑定端口号
server_socket.bind((host, port))

# 设置最大连接数，超过后排队
server_socket.listen(5)

while True:
    # 建立客户端连接
    client_socket, addr = server_socket.accept()
    print(f"连接地址: {addr}")

    # 接收数据
    data = client_socket.recv(1024)
    print(f"接收数据: {data.decode('utf-8')}")

    # 发送数据
    message = "Hello, client!"
    client_socket.send(message.encode('utf-8'))

    # 关闭连接
    client_socket.close()

性能优化

多线程处理多个客户端连接：

import socket
import threading


def handle_client(client_socket, addr):
    print(f"连接地址: {addr}")
    data = client_socket.recv(1024)
    print(f"接收数据: {data.decode('utf-8')}")
    message = "Hello, client!"
    client_socket.send(message.encode('utf-8'))
    client_socket.close()


server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
host = socket.gethostname()
port = 9999
server_socket.bind((host, port))
server_socket.listen(5)

while True:
    client_socket, addr = server_socket.accept()
    client_thread = threading.Thread(target=handle_client, args=(client_socket, addr))
    client_thread.start()

这种方式为每个客户端连接创建一个新线程，从而可以同时处理多个客户端请求。不过，多线程有线程切换开销和 GIL（全局解释器锁）的限制，在 CPU 密集型任务时性能提升有限。

使用 select 或 epoll 多路复用（适用于 Linux 系统）：

import socket
import select

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
host = socket.gethostname()
port = 9999
server_socket.bind((host, port))
server_socket.listen(5)

inputs = [server_socket]

while True:
    readable, _, _ = select.select(inputs, [], [])
    for sock in readable:
        if sock is server_socket:
            client_socket, addr = server_socket.accept()
            inputs.append(client_socket)
        else:
            data = sock.recv(1024)
            if data:
                print(f"接收数据: {data.decode('utf-8')}")
                message = "Hello, client!"
                sock.send(message.encode('utf-8'))
            else:
                inputs.remove(sock)
                sock.close()

select 可以监控多个文件描述符（这里指 socket），当有事件发生（例如可读、可写等）时通知程序，这样可以在单线程内处理多个客户端连接，减少线程切换开销。epoll 是 select 的改进版本，在处理大量连接时性能更优，不过 epoll 是 Linux 特有的，select 是跨平台的。

异步 I/O（使用 asyncio 库）：

import asyncio


async def handle_connection(reader, writer):
    data = await reader.read(1024)
    print(f"接收数据: {data.decode('utf-8')}")
    message = "Hello, client!"
    writer.write(message.encode('utf-8'))
    await writer.drain()
    writer.close()


async def main():
    server = await asyncio.start_server(handle_connection, 'localhost', 9999)
    async with server:
        await server.serve_forever()


asyncio.run(main())

asyncio 提供了基于协程的异步 I/O 支持，在单线程内通过事件循环高效处理多个 I/O 操作，非常适合处理大量并发连接且 I/O 密集型的场景。

星途面试题库

面试题：Python 中如何使用 socket 模块创建一个基本的高性能 TCP 服务器框架

知识考点

面试题答案

TCP 服务器示例代码

性能优化