本文目录导读：
我来用一个简单的多人在线聊天室案例演示 socketserver 模块的用法，这个案例包含服务端和客户端，支持多用户同时在线聊天。
服务端代码 (server.py)
import socketserver
import threading
from datetime import datetime
class ChatHandler(socketserver.StreamRequestHandler):
# 存储所有连接的客户端
clients = {}
clients_lock = threading.Lock()
def setup(self):
"""连接建立时的初始化"""
print(f"新客户端连接: {self.client_address}")
def handle(self):
"""处理客户端消息的主循环"""
# 客户端连接后，先接收用户名
try:
username = self.rfile.readline().strip().decode('utf-8')
if not username:
return
# 将客户端加入在线列表
with self.clients_lock:
self.__class__.clients[username] = self
# 广播用户上线消息
self.broadcast(f"📢 {username} 加入了聊天室", exclude=None)
# 发送欢迎消息给当前用户
welcome_msg = f"🟢 欢迎 {username} 加入聊天室！当前在线人数: {len(self.clients)}"
self.send_message(welcome_msg)
self.send_message("💡 输入 'quit' 退出聊天, 输入 'users' 查看在线用户")
# 持续接收和转发消息
while True:
data = self.rfile.readline().strip()
if not data:
break
message = data.decode('utf-8')
# 处理特殊命令
if message.lower() == 'quit':
break
elif message.lower() == 'users':
# 显示在线用户
with self.clients_lock:
online_users = ', '.join(self.clients.keys())
self.send_message(f"👥 当前在线用户: {online_users}")
else:
# 广播普通消息
timestamp = datetime.now().strftime("%H:%M:%S")
formatted_msg = f"[{timestamp}] {username}: {message}"
self.broadcast(formatted_msg, exclude=None)
except Exception as e:
print(f"处理客户端消息时出错: {e}")
finally:
self.cleanup(username)
def cleanup(self, username):
"""清理客户端连接"""
if username:
with self.clients_lock:
if username in self.clients:
del self.clients[username]
# 广播用户离线消息
self.broadcast(f"📢 {username} 离开了聊天室", exclude=None)
print(f"客户端断开: {username} ({self.client_address})")
def send_message(self, message):
"""发送消息到当前客户端"""
try:
self.wfile.write((message + '\n').encode('utf-8'))
self.wfile.flush()
except:
pass
def broadcast(self, message, exclude=None):
"""广播消息给所有客户端（可选排除某个客户端）"""
with self.clients_lock:
for username, client in self.clients.items():
if username != exclude:
try:
client.send_message(message)
except:
pass
class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
"""支持多线程的TCP服务器"""
allow_reuse_address = True
daemon_threads = True
def main():
HOST = 'localhost'
PORT = 12345
server = ThreadedTCPServer((HOST, PORT), ChatHandler)
print(f"服务器启动在 {HOST}:{PORT}")
print("等待客户端连接...")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n服务器关闭...")
server.shutdown()
server.server_close()
if __name__ == '__main__':
main()
客户端代码 (client.py)
import socket
import threading
import sys
class ChatClient:
def __init__(self, host='localhost', port=12345):
self.host = host
self.port = port
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.running = True
def connect(self):
"""连接到服务器"""
try:
self.socket.connect((self.host, self.port))
print(f"已连接到服务器 {self.host}:{self.port}")
return True
except Exception as e:
print(f"连接失败: {e}")
return False
def send_username(self, username):
"""发送用户名"""
self.socket.send((username + '\n').encode('utf-8'))
def receive_messages(self):
"""接收消息的线程"""
while self.running:
try:
data = self.socket.recv(1024)
if not data:
print("\n[连接已断开]")
self.running = False
break
message = data.decode('utf-8').strip()
if message:
print(f"\r{message}\n> ", end='', flush=True)
except Exception as e:
if self.running:
print(f"\n[接收消息出错: {e}]")
self.running = False
break
def send_messages(self):
"""发送消息的主循环"""
while self.running:
try:
message = input("> ")
if message:
self.socket.send((message + '\n').encode('utf-8'))
if message.lower() == 'quit':
break
except (EOFError, KeyboardInterrupt):
break
except Exception as e:
print(f"发送消息出错: {e}")
break
self.running = False
self.socket.close()
def main():
# 获取用户输入
host = input("服务器地址 (默认 localhost): ") or 'localhost'
port = input("端口号 (默认 12345): ") or '12345'
username = input("请输入你的昵称: ")
# 创建并连接客户端
client = ChatClient(host, int(port))
if not client.connect():
return
# 发送用户名
client.send_username(username)
print("\n=== 欢迎来到聊天室 ===")
print("输入消息发送，输入 'quit' 退出")
print("=" * 30)
# 启动接收线程
receive_thread = threading.Thread(target=client.receive_messages)
receive_thread.daemon = True
receive_thread.start()
# 在主线程中发送消息
client.send_messages()
print("\n已退出聊天室")
if __name__ == '__main__':
main()
运行效果演示
第一步：启动服务器
python server.py # 输出： # 服务器启动在 localhost:12345 # 等待客户端连接...
第二步：启动多个客户端（打开多个终端）
# 终端1 - 用户 Alice python client.py # 输入：服务器地址 localhost # 输入：端口 12345 # 输入：昵称 Alice # 终端2 - 用户 Bob python client.py # 输入：服务器地址 localhost # 输入：端口 12345 # 输入：昵称 Bob
聊天演示
# Alice 的终端：
> 大家好！
[14:30:25] Alice: 大家好！
> 今天天气不错
[14:30:45] Alice: 今天天气不错
# Bob 的终端：
📢 Alice 加入了聊天室
> 嗨，Alice！
[14:30:25] Alice: 大家好！
>
[14:30:45] Alice: 今天天气不错
> 是啊，很适合出去玩
[14:31:00] Bob: 是啊，很适合出去玩
# 使用命令：
> users
👥 当前在线用户: Alice, Bob
> quit
📢 Bob 离开了聊天室关键知识点
socketserver 核心类
StreamRequestHandler：处理TCP连接ThreadingMixIn：添加多线程支持TCPServer：基础的TCP服务器
主要方法
setup()：连接初始化handle()：处理客户端请求finish()：连接清理
通信协议
- 使用
rfile.readline()读取数据（文本模式） - 使用
wfile.write()发送数据 - 每条消息以换行符
\n
线程安全
- 使用
threading.Lock保护共享数据 daemon_threads = True自动管理线程生命周期
这个案例展示了 socketserver 模块的核心用法，包括多线程处理、消息广播、用户管理等实际聊天室功能，你可以基于此扩展更多功能，如私聊、文件传输、房间管理等。