如何通过FastAPI WebSocket实现消息已读状态查询功能？

核心思路 #

要实现类似WhatsApp的已读/未读状态，需要完成以下关键环节：

• 给每条消息分配唯一标识，追踪其状态（未读/已读）
• 客户端收到消息后主动发送已读确认
• 服务端更新消息状态并通知原发送者
• 前端根据状态展示对应标识

代码实现步骤 #

1. 定义消息状态与数据结构 #

首先定义消息状态枚举和消息模型，规范消息的存储和状态追踪：

from enum import Enum
import uuid
from typing import Dict, List, Optional
from fastapi import WebSocket, HTMLResponse

class MessageStatus(str, Enum):
    UNREAD = "unread"
    READ = "read"

class Message:
    def __init__(self, message_id: str, sender_id: int, receiver_id: int, content: str):
        self.message_id = message_id
        self.sender_id = sender_id
        self.receiver_id = receiver_id
        self.content = content
        self.status = MessageStatus.UNREAD

2. 重构连接管理器 #

修改ConnectionManager，增加客户端ID与连接的映射，以及消息存储功能：

class ConnectionManager:
    def __init__(self):
        self.active_connections: Dict[int, WebSocket] = {}  # 存储客户端ID到WebSocket的映射
        self.messages: List[Message] = []  # 内存存储消息（实际项目建议用数据库）

    async def connect(self, websocket: WebSocket, client_id: int):
        await websocket.accept()
        self.active_connections[client_id] = websocket

    def disconnect(self, client_id: int):
        if client_id in self.active_connections:
            del self.active_connections[client_id]

    async def send_personal_message(self, data: dict, client_id: int):
        websocket = self.active_connections.get(client_id)
        if websocket:
            await websocket.send_json(data)

    def add_message(self, sender_id: int, receiver_id: int, content: str) -> Message:
        # 生成唯一消息ID
        message_id = str(uuid.uuid4())
        message = Message(message_id, sender_id, receiver_id, content)
        self.messages.append(message)
        return message

    def mark_message_as_read(self, message_id: str) -> Optional[Message]:
        # 标记消息为已读
        for message in self.messages:
            if message.message_id == message_id:
                message.status = MessageStatus.READ
                return message
        return None

manager = ConnectionManager()

3. 修改WebSocket端点逻辑 #

更新WebSocket处理函数，支持普通消息发送和已读确认的处理：

@app.get("/")
async def get():
    return HTMLResponse(html)  # 这里的html是你的前端页面代码

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket, client_id)
    try:
        while True:
            data = await websocket.receive_json()
            # 处理普通消息发送
            if data.get("type") == "message":
                receiver_id = data["receiver_id"]
                content = data["content"]
                # 保存消息并获取实例
                message = manager.add_message(client_id, receiver_id, content)
                # 发送消息给接收者
                await manager.send_personal_message({
                    "type": "new_message",
                    "message_id": message.message_id,
                    "sender_id": message.sender_id,
                    "content": message.content
                }, receiver_id)
                # 回复发送者消息已发出，状态为未读
                await manager.send_personal_message({
                    "type": "message_sent",
                    "message_id": message.message_id,
                    "status": message.status
                }, client_id)
            # 处理已读确认
            elif data.get("type") == "read_confirmation":
                message_id = data["message_id"]
                message = manager.mark_message_as_read(message_id)
                if message:
                    # 通知原发送者消息已读
                    await manager.send_personal_message({
                        "type": "message_read",
                        "message_id": message.message_id,
                        "status": message.status
                    }, message.sender_id)
    except WebSocketDisconnect:
        manager.disconnect(client_id)
        await manager.broadcast(f"Client #{client_id} left the chat")

4. 前端交互逻辑 #

前端需要处理消息接收、发送已读确认，并根据状态更新界面：

// 替换为当前用户的ID
const currentClientId = 1;
const ws = new WebSocket(`ws://localhost:8000/ws/${currentClientId}`);

// 接收服务端消息
ws.onmessage = function(event) {
    const data = JSON.parse(event.data);
    switch(data.type) {
        case "new_message":
            // 在界面上渲染新消息
            renderMessage(data);
            // 发送已读确认给服务端
            ws.send(JSON.stringify({
                type: "read_confirmation",
                message_id: data.message_id
            }));
            break;
        case "message_sent":
            // 更新发送消息的初始状态（未读）
            updateMessageStatus(data.message_id, data.status);
            break;
        case "message_read":
            // 更新发送消息的状态为已读
            updateMessageStatus(data.message_id, data.status);
            break;
    }
};

// 发送消息到指定用户
function sendMessage(receiverId, content) {
    ws.send(JSON.stringify({
        type: "message",
        receiver_id: receiverId,
        content: content
    }));
}

// 更新消息状态展示（类似WhatsApp的对勾）
function updateMessageStatus(messageId, status) {
    const messageElement = document.getElementById(`msg-${messageId}`);
    if (!messageElement) return;
    
    const statusEl = messageElement.querySelector(".msg-status");
    if (status === "read") {
        statusEl.textContent = "✓✓";
        statusEl.style.color = "#25D366"; // WhatsApp蓝绿色
    } else {
        statusEl.textContent = "✓";
        statusEl.style.color = "#999";
    }
}

// 渲染消息到界面
function renderMessage(message) {
    const chatContainer = document.getElementById("chat-container");
    const msgDiv = document.createElement("div");
    msgDiv.id = `msg-${message.message_id}`;
    msgDiv.className = message.sender_id === currentClientId ? "sent-msg" : "received-msg";
    msgDiv.innerHTML = `
        <div class="msg-content">${message.content}</div>
        <span class="msg-status">✓</span>
    `;
    chatContainer.appendChild(msgDiv);
}

实际项目优化建议 #

• 持久化存储：内存存储消息重启后会丢失，建议用SQLAlchemy连接MySQL/PostgreSQL，或者MongoDB存储消息记录
• 离线消息处理：当接收者离线时，消息标记为未读，用户上线后主动同步未读消息
• 批量已读确认：如果一次性接收多条消息，可以批量发送已读确认，减少WebSocket通信次数
• 状态同步：用户重新连接时，拉取自己发送的消息的最新状态

内容的提问来源于stack exchange，提问作者RunCoderRun