当前位置：首页 > news >正文

完全用python 实现消息中间件4

news 2025/7/21 20:11:23

为了进一步完善这个消息中间件，我们可以添加以下功能：

消息确认：客户端可以发送一个确认消息，表明消息已经被正确接收。
消息队列：使用一个队列来存储消息，而不是直接存储在字典中。
多消费者支持：允许多个消费者同时消费消息。
消息索引：提供一个接口来检索特定消息。
消息过期：自动删除过期的消息。
消息状态：为每个消息添加状态，如发送、接收、确认等。
错误处理：实现基本的错误处理和重试机制。
由于我们不使用外部存储系统，这些功能将完全在内存中实现。以下是更新后的代码：

from fastapi import FastAPI, HTTPException, Depends, status
from typing import Dict, List, Optional
import json
from threading import Lock
from uuid import uuid4
app = FastAPI()
# 存储消息的字典，键为频道名，值为消息队列
channels: Dict[str, List[Dict[str, str]]] = {}
# 消息锁，用于并发控制
lock = Lock()
# 消息结构示例
message_example = {"id": "message_id","content": "Hello, World!","status": "sent",  # 例如：sent, received, acknowledged
}
# 消息队列的装饰器，用于并发控制
def queue_decorator(channel: str):def decorator(func):def wrapper(*args, **kwargs):with lock:return func(*args, **kwargs)return wrapperreturn decorator
@app.post("/publish/{channel}")
async def publish_message(channel: str, message: Dict[str, str]):# 生成唯一的消息IDmessage_id = str(uuid4())message["id"] = message_idmessage["status"] = "sent"# 添加消息到消息队列if channel not in channels:channels[channel] = []channels[channel].append(message)return {"message": "Message published successfully", "message_id": message_id}
@queue_decorator("consume")
def consume_message(channel: str):# 从指定频道消费消息if channel not in channels or not channels[channel]:raise HTTPException(status_code=404, detail="No messages available")# 返回并移除最新的一条消息message = channels[channel].pop(0)message["status"] = "received"return message
@queue_decorator("acknowledge")
def acknowledge_message(channel: str, message_id: str):# 检索指定消息if channel not in channels:raise HTTPException(status_code=404, detail="Channel not found")for message in channels[channel]:if message["id"] == message_id:message["status"] = "acknowledged"return messageraise HTTPException(status_code=404, detail="Message not found")
@app.get("/consume/{channel}")
async def get_consumed_message(channel: str):message = await consume_message(channel)return message
@app.get("/acknowledge/{channel}/{message_id}")
async def get_acknowledged_message(channel: str, message_id: str):message = await acknowledge_message(channel, message_id)return message
if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)