Python WebSocket 实时双向通信服务

🖥️ 运行效果

🚀 实例简介

这是一个基于 Python 的 WebSocket 实时双向通信演示程序，使用 tkinter 构建图形界面。程序内置 WebSocket 服务器和客户端，可以在同一个应用中演示完整的实时通信流程。支持多用户连接、实时消息广播、服务器日志显示等功能，是学习 WebSocket 协议和异步编程的优秀示例。

💻 核心代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Python WebSocket 实时双向通信服务
===================================
功能：内置 WebSocket 服务器 + 客户端，演示实时双向通信
技术：websockets 库 + asyncio + tkinter GUI
作者：好例子网 (haolizi.com)
"""

import tkinter as tk
from tkinter import ttk, scrolledtext
import threading
import asyncio
import json
import time
import platform
from datetime import datetime

# 中文字体配置（无头环境必须用元组形式）
CHINESE_FONT = "WenQuanYi Zen Hei" if platform.system() == "Linux" else "Microsoft YaHei"

try:
    import websockets
    import websockets.server
    import websockets.client
except ImportError:
    print("请先安装 websockets: pip install websockets")
    exit(1)


class WebSocketServer:
    """内置 WebSocket 服务器（后台线程运行）"""

    def __init__(self, host="localhost", port=8765, on_message_callback=None):
        self.host = host
        self.port = port
        self.clients = set()
        self.on_message_callback = on_message_callback
        self.loop = None
        self.server = None
        self.running = False

    def start(self):
        """在独立线程中启动服务器"""
        self.running = True
        thread = threading.Thread(target=self._run_server, daemon=True)
        thread.start()
        time.sleep(0.5)

    def _run_server(self):
        """服务器主循环（asyncio 事件循环）"""
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        try:
            self.loop.run_until_complete(self._serve())
        except Exception as e:
            if self.on_message_callback:
                self.on_message_callback(f"[服务器错误] {e}")

    async def _serve(self):
        """异步服务器逻辑"""
        self.server = await websockets.serve(self._handler, self.host, self.port)
        if self.on_message_callback:
            self.on_message_callback(f"[服务器] 已启动 ws://{self.host}:{self.port}")
        await self.server.wait_closed()

    async def _handler(self, websocket):
        """处理每个客户端连接"""
        self.clients.add(websocket)
        remote = websocket.remote_address
        if self.on_message_callback:
            self.on_message_callback(f"[服务器] 客户端已连接: {remote}")
        try:
            async for message in websocket:
                data = json.loads(message)
                content = data.get("content", "")
                sender = data.get("sender", "匿名")
                timestamp = datetime.now().strftime("%H:%M:%S")

💡 适用场景

办公自动化、数据分析、快速原型开发