我来给你一个完整的实现方案。核心思路是：用Redis的Pub/Sub做消息分发，避免单连接阻塞；用字典存储WebSocket连接，实现用户管理。

import tornado.ioloop
import tornado.web
import tornado.websocket
import asyncio
import redis.asyncio as redis
import json
import uuid

class WebSocketHandler(tornado.websocket.WebSocketHandler):
    # 存储所有活跃连接 {connection_id: WebSocketHandler实例}
    connections = {}
    
    def initialize(self, redis_client):
        self.redis_client = redis_client
        self.connection_id = str(uuid.uuid4())
        self.pubsub = None
        
    async def open(self):
        """客户端连接时调用"""
        # 存储连接
        WebSocketHandler.connections[self.connection_id] = self
        print(f"新连接: {self.connection_id}, 当前连接数: {len(WebSocketHandler.connections)}")
        
        # 订阅Redis频道
        self.pubsub = self.redis_client.pubsub()
        await self.pubsub.subscribe('broadcast_channel')
        
        # 启动消息监听任务
        asyncio.create_task(self.listen_redis())
        
    async def listen_redis(self):
        """监听Redis消息并推送给客户端"""
        try:
            async for message in self.pubsub.listen():
                if message['type'] == 'message':
                    data = message['data']
                    if isinstance(data, bytes):
                        data = data.decode('utf-8')
                    
                    # 推送给当前客户端
                    if self.ws_connection:
                        await self.write_message(data)
        except Exception as e:
            print(f"监听Redis出错: {e}")
        finally:
            if self.pubsub:
                await self.pubsub.unsubscribe()
                await self.pubsub.close()
    
    async def on_message(self, message):
        """接收客户端消息"""
        try:
            data = json.loads(message)
            # 这里可以处理客户端发送的消息
            # 例如：{"type": "chat", "content": "hello"}
            print(f"收到消息: {data}")
            
            # 如果需要广播给所有客户端，可以发布到Redis
            # await self.redis_client.publish('broadcast_channel', json.dumps(data))
            
        except json.JSONDecodeError:
            print(f"无效的JSON: {message}")
    
    def on_close(self):
        """客户端断开连接时调用"""
        if self.connection_id in WebSocketHandler.connections:
            del WebSocketHandler.connections[self.connection_id]
        print(f"连接关闭: {self.connection_id}, 剩余连接数: {len(WebSocketHandler.connections)}")
        
        # 关闭Redis订阅
        if self.pubsub:
            asyncio.create_task(self.pubsub.unsubscribe())

class BroadcastHandler(tornado.web.RequestHandler):
    """HTTP接口，用于触发广播"""
    
    def initialize(self, redis_client):
        self.redis_client = redis_client
    
    async def post(self):
        """接收广播消息并发布到Redis"""
        try:
            message = self.get_argument('message', '')
            if not message:
                self.write({'error': '消息不能为空'})
                return
                
            # 发布到Redis频道
            await self.redis_client.publish('broadcast_channel', message)
            
            self.write({'success': True, 'message': '广播已发送'})
        except Exception as e:
            self.write({'error': str(e)})

async def make_app():
    """创建应用"""
    # 创建Redis客户端
    redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
    
    # 测试Redis连接
    try:
        await redis_client.ping()
        print("Redis连接成功")
    except Exception as e:
        print(f"Redis连接失败: {e}")
        raise
    
    return tornado.web.Application([
        (r'/ws', WebSocketHandler, {'redis_client': redis_client}),
        (r'/broadcast', BroadcastHandler, {'redis_client': redis_client}),
    ])

async def main():
    """主函数"""
    app = await make_app()
    app.listen(8888)
    print("WebSocket服务器启动在: ws://localhost:8888/ws")
    print("广播接口: POST http://localhost:8888/broadcast?message=你的消息")
    
    # 启动Tornado事件循环
    await asyncio.Event().wait()

if __name__ == '__main__':
    asyncio.run(main())

关键点解释：

1. Redis Pub/Sub解耦：每个WebSocket连接独立订阅Redis频道，消息通过Redis广播，避免单连接阻塞影响其他客户端。

2. 连接管理：用类变量connections字典存储所有活跃连接，key是唯一的connection_id，value是WebSocketHandler实例。

3. 异步处理：使用async/await确保非阻塞，listen_redis()在后台任务中运行，持续监听Redis消息。

4. 消息推送：收到Redis消息后，通过write_message()推送给对应客户端。

5. 资源清理：连接关闭时，从connections字典移除，并取消Redis订阅。

使用方式：

1. 启动Redis服务器
2. 运行Python脚本
3. 客户端连接：ws://localhost:8888/ws
4. 发送广播：POST请求到http://localhost:8888/broadcast?message=你的消息

总结：用Redis做消息中转，连接字典做用户管理，异步避免阻塞。