WebSocket plays an increasingly important role in modern web development, especially in scenarios requiring real-time communication. As a leading asynchronous web framework in Python, aiohttp performs exceptionally well in handling WebSocket connections. Today, we will delve into how to build high-performance WebSocket applications with aiohttp, from basic setup to performance tuning, step by step creating your own real-time communication tool.

Basic WebSocket Server Setup

Setting up a WebSocket server is not as complicated as it seems. Let’s take a look at the simplest example:

from aiohttp import web, WSMsgType
import asyncio
import logging
# Store all connected websockets
connections = set()
async def websocket_handler(request):

ws = web.WebSocketResponse() await ws.prepare(request)

Add New Connection to Set

connections.add(ws) print(f”New connection established, current connection count: {len(connections)}”)

try: async for msg in ws: if msg.type == WSMsgType.TEXT: data = msg.data print(f”Received message: {data}”)

        # Broadcast to all connections
        await broadcast_message(data)
    elif msg.type == WSMsgType.ERROR:
        print(f"WebSocket error: {ws.exception()}")

finally: # Clean up when connection is closed connections.discard(ws) print(f”Connection closed, remaining connections: {len(connections)}”)

return ws

async def broadcast_message(message):

“””Broadcast message to all connections””” if not connections: return

Create Task List and Send Messages Concurrently

tasks = [] for ws in connections.copy(): # Copy set to avoid modification during iteration if ws.closed: connections.discard(ws) else: tasks.append(ws.send_str(message))

if tasks: await asyncio.gather(*tasks, return_exceptions=True)

app = web.Application()
app.router.add_get('/ws', websocket_handler)
if __name__ == '__main__':

web.run_app(app, host=’localhost’, port=8080)

This basic version can already handle WebSocket connections, but there is still much room for optimization.

⚠️ Tip: Use <span>connections.copy()</span> to avoid exceptions caused by modifying the set during iteration. WebSocket connections can be disconnected at any time, so always check the connection status.

Connection Management and Message Handling Optimization

In real projects, connection management is not as simple as storing a set. We need more precise control:

import json
import time
from collections import defaultdict
from dataclasses import dataclass
from typing import Dict, Set, Optional
@dataclass
class Connection:

websocket: web.WebSocketResponse user_id: str room_id: str connected_at: float last_ping: float = 0

class ConnectionManager:

def __init__(self): # Manage connections grouped by room self.rooms: Dict[str, Set[Connection]] = defaultdict(set) # Index connections by user ID self.user_connections: Dict[str, Connection] = {}

async def connect(self, websocket: web.WebSocketResponse, user_id: str, room_id: str = “default”) -> Connection: “””Establish a new connection””” # If the user already has a connection, disconnect the old one first await self.disconnect_user(user_id)

conn = Connection(
    websocket=websocket,
    user_id=user_id,
    room_id=room_id,
    connected_at=time.time()
)
self.rooms[room_id].add(conn)
self.user_connections[user_id] = conn
return conn

async def disconnect_user(self, user_id: str): “””Disconnect specified user’s connection””” if user_id in self.user_connections: conn = self.user_connections[user_id] self.rooms[conn.room_id].discard(conn) del self.user_connections[user_id]

    if not conn.websocket.closed:
        await conn.websocket.close()

async def send_to_room(self, room_id: str, message: dict, exclude_user: Optional[str] = None): “””Send message to all users in the room””” room_connections = self.rooms.get(room_id, set()) if not room_connections: return

message_str = json.dumps(message, ensure_ascii=False)
tasks = []
for conn in room_connections.copy():
    if exclude_user and conn.user_id == exclude_user:
        continue
    if conn.websocket.closed:
        self.rooms[room_id].discard(conn)
        self.user_connections.pop(conn.user_id, None)
    else:
        tasks.append(self._safe_send(conn.websocket, message_str))
if tasks:
    await asyncio.gather(*tasks, return_exceptions=True)

async def _safe_send(self, ws: web.WebSocketResponse, message: str): “””Safely send message, catch exceptions””” try: await ws.send_str(message) except ConnectionResetError: # Connection has been closed, ignore error pass except Exception as e: print(f”Failed to send message: {e}”)

# Global connection manager
connection_manager = ConnectionManager()

Now our connection management is much more professional, supporting the concept of rooms, user indexing, and elegantly handling disconnections.

Core Techniques for Performance Optimization

There are several key points for WebSocket performance optimization, mastering them can significantly increase your application’s handling capacity:

Batch Message Processing

import asyncio
from collections import deque
class MessageBuffer:

def __init__(self, max_size: int = 100, flush_interval: float = 0.1): self.buffer = deque(maxlen=max_size) self.max_size = max_size self.flush_interval = flush_interval self._flush_task = None

async def add_message(self, room_id: str, message: dict): “””Add message to buffer””” self.buffer.append((room_id, message))

# Immediately flush if buffer is full
if len(self.buffer) >= self.max_size:
    await self._flush()
elif not self._flush_task:
    # Start timed flush task
    self._flush_task = asyncio.create_task(self._delayed_flush())

async def _delayed_flush(self): “””Delayed flush””” await asyncio.sleep(self.flush_interval) await self._flush()

async def _flush(self): “””Flush buffer””” if not self.buffer: return

# Group messages by room
room_messages = defaultdict(list)
while self.buffer:
    room_id, message = self.buffer.popleft()
    room_messages[room_id].append(message)
# Batch send
tasks = []
for room_id, messages in room_messages.items():
    batch_message = {
        "type": "batch",
        "messages": messages,
        "timestamp": time.time()
    }
    tasks.append(connection_manager.send_to_room(room_id, batch_message))
if tasks:
    await asyncio.gather(*tasks, return_exceptions=True)
self._flush_task = None

# Global message buffer
message_buffer = MessageBuffer()

Heartbeat Detection Mechanism

async def heartbeat_checker():

“””Heartbeat detection task””” while True: current_time = time.time() timeout_users = []

for user_id, conn in connection_manager.user_connections.items():
    # If no heartbeat for more than 30 seconds, consider it disconnected
    if current_time - conn.last_ping > 30:
        timeout_users.append(user_id)
# Clean up timed out connections
for user_id in timeout_users:
    print(f"User {user_id} heartbeat timed out, disconnecting")
    await connection_manager.disconnect_user(user_id)
await asyncio.sleep(10)  # Check every 10 seconds

# Start heartbeat checker when app starts
async def init_app():

asyncio.create_task(heartbeat_checker())

⚠️ Tip: Batch message processing can significantly improve performance, especially in high-concurrency scenarios. However, be mindful of the balance between batch size and delay time; too large can consume memory, while too small may not achieve optimization.

Practical Case: High-Performance Chat Room

Integrating the techniques learned earlier, let’s build a complete chat room:

async def chat_websocket_handler(request):

“””Chat room WebSocket handler””” ws = web.WebSocketResponse(heartbeat=30) # 30 seconds heartbeat await ws.prepare(request)

user_id = request.query.get(‘user_id’) room_id = request.query.get(‘room_id’, ‘default’)

If not user_id: await ws.close(code=4000, message=b’Missing user_id’) return ws

Establish Connection

conn = await connection_manager.connect(ws, user_id, room_id)

Send Welcome Message

welcome_msg = { “type”: “system”, “message”: f”User {user_id} has joined the room”, “timestamp”: time.time() } await message_buffer.add_message(room_id, welcome_msg)

try: async for msg in ws: if msg.type == WSMsgType.TEXT: try: data = json.loads(msg.data) await handle_message(conn, data) except json.JSONDecodeError: await ws.send_str(json.dumps({ “type”: “error”, “message”: “Message format error” })) elif msg.type == WSMsgType.PONG: # Update heartbeat time conn.last_ping = time.time() elif msg.type == WSMsgType.ERROR: print(f”WebSocket error: {ws.exception()}”)

except asyncio.CancelledError: pass finally: # Send leave message leave_msg = { “type”: “system”, “message”: f”User {user_id} has left the room”, “timestamp”: time.time() } await message_buffer.add_message(room_id, leave_msg) await connection_manager.disconnect_user(user_id)

return ws

async def handle_message(conn: Connection, data: dict):

“””Handle received messages””” msg_type = data.get(‘type’)

If msg_type == ‘chat’: # Chat message chat_msg = { “type”: “chat”, “user_id”: conn.user_id, “message”: data.get(‘message’, ”), “timestamp”: time.time() } await message_buffer.add_message(conn.room_id, chat_msg)

elif msg_type == ‘ping’: # Heartbeat message conn.last_ping = time.time() await conn.websocket.send_str(json.dumps({ “type”: “pong”, “timestamp”: time.time() }))

This chat room already meets the basic requirements for a production environment: supporting multiple rooms, heartbeat detection, message buffering, error handling, etc.

Avoid Common Pitfalls

WebSocket development has many pitfalls; here are a few I have encountered:

Memory Leak Issues: Failing to clean up references promptly after disconnection leads to continuous memory growth. The solution is to ensure all references are cleaned up in the <span>finally</span> block.

Concurrency Safety Issues: Multiple coroutines modifying the connection set simultaneously may cause exceptions. Use <span>asyncio.Lock</span> to protect critical sections:

class ThreadSafeConnectionManager:

def __init__(self): self.rooms = defaultdict(set) self.user_connections = {} self._lock = asyncio.Lock()

async def connect(self, websocket, user_id, room_id): async with self._lock: # Connection logic pass

Message Loss Issues: Messages may be lost during network instability. You can add a message acknowledgment mechanism:

async def send_with_ack(ws, message_id, content):

“””Send message requiring acknowledgment””” message = { “id”: message_id, “content”: content, “require_ack”: True } await ws.send_str(json.dumps(message))

You can use Redis to store messages awaiting acknowledgment and resend them on timeout

⚠️ Tip: In production, remember to configure appropriate WebSocket timeout settings; too short may kill normal connections, while too long may consume resources. Typically, 30-60 seconds is appropriate.

Performance Monitoring and Debugging

Want to know how your WebSocket service is performing? Add some monitoring code:

import psutil
from datetime import datetime
class PerformanceMonitor:

def __init__(self): self.start_time = time.time() self.message_count = 0

def log_stats(self): “””Output performance statistics””” uptime = time.time() – self.start_time memory_usage = psutil.virtual_memory().percent cpu_usage = psutil.cpu_percent()

stats = {
    "uptime": f"{uptime:.1f}s",
    "connections": len(connection_manager.user_connections),
    "messages_processed": self.message_count,
    "messages_per_second": self.message_count / uptime if uptime > 0 else 0,
    "memory_usage": f"{memory_usage:.1f}%",
    "cpu_usage": f"{cpu_usage:.1f}%"
}
print(f"[{datetime.now()}] Stats: {stats}")

# Output statistics every minute
monitor = PerformanceMonitor()
async def stats_reporter:

aiohttp’s WebSocket support is already quite mature. By properly using these optimization techniques, your real-time applications can easily handle high-concurrency scenarios. Remember, performance optimization is a gradual process; first, stabilize the basic functionality, then gradually add optimization measures.

Now that you have mastered the core technologies of aiohttp WebSocket, you can start building your own real-time communication applications. There is no silver bullet for performance optimization; choose the right strategy based on specific scenarios, and continuous testing and adjustment are necessary to find the best solution.