如何在aiogram-3.x（beta）中实现类似aiogram-2的Anti-Flood中间件？

完全可以实现类似aiogram 2.x的Anti-Flood中间件，核心逻辑是通过记录用户上次发送消息的时间，判断是否超过设定的频率阈值，以下是具体实现方案：

1. 定义Anti-Flood中间件类 #

基于aiogram 3.x的BaseMiddleware构建，用字典存储用户最后发送消息的时间戳：

from aiogram import BaseMiddleware
from aiogram.types import Message
from datetime import datetime, timedelta
from typing import Callable, Dict, Any, Awaitable

class AntiFloodMiddleware(BaseMiddleware):
    def __init__(self, rate_limit: int):
        # 限流阈值（秒）
        self.rate_limit = rate_limit
        # 存储用户ID与最后发送时间的映射
        self.user_last_send: Dict[int, datetime] = {}

    async def __call__(
        self,
        handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
        event: Message,
        data: Dict[str, Any]
    ) -> Any:
        user_id = event.from_user.id
        current_time = datetime.now()

        # 检查是否触发限流
        if user_id in self.user_last_send:
            time_diff = current_time - self.user_last_send[user_id]
            if time_diff < timedelta(seconds=self.rate_limit):
                await event.answer("发送过于频繁，请稍后再试！")
                return None  # 终止后续处理流程

        # 更新用户最后发送时间
        self.user_last_send[user_id] = current_time
        # 继续执行后续处理器
        return await handler(event, data)

2. 注册中间件 #

aiogram 3.x支持全局注册或针对特定路由器注册，按需选择：

全局注册（所有消息生效） #

from aiogram import Bot, Dispatcher
from aiogram.filters import CommandStart

# 初始化Bot和Dispatcher
bot = Bot(token="你的机器人Token")
dp = Dispatcher()

# 挂载全局消息中间件，设置1秒限流阈值
dp.message.middleware(AntiFloodMiddleware(rate_limit=1))

# 示例命令处理器
@dp.message(CommandStart())
async def handle_start(message: Message):
    await message.answer("欢迎使用机器人！")

# 启动Polling
async def main():
    await dp.start_polling(bot)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

针对特定路由器注册 #

如果只需要对部分路由的消息限流，可以单独给路由器挂载中间件：

from aiogram import Router

# 创建独立路由器
content_router = Router()
# 给该路由器设置2秒限流阈值
content_router.message.middleware(AntiFloodMiddleware(rate_limit=2))

# 该路由器下的消息处理器
@content_router.message()
async def handle_content(message: Message):
    await message.answer("这条消息受限流保护")

# 将路由器挂载到Dispatcher
dp.include_router(content_router)

扩展说明 #

• 若需要持久化存储用户发送记录（避免重启机器人后丢失），可以用Redis等缓存替代字典
• 可根据用户权限设置不同限流阈值，比如管理员不受限制
• 超过阈值后除了发送提示，还可以扩展临时禁言等更严格的限制逻辑

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