如何让基于Aiogram 2.3的Telegram机器人支持多用户同时使用？

你的核心问题并非Bot实例单例导致，Aiogram本身就是基于单例Dispatcher处理所有请求的。混乱的根源是你自己维护的全局变量、状态未按用户维度隔离，而OOP的核心思路是为每个用户创建独立的会话对象，存储其专属数据。

解决方案实现步骤 #

1. 定义用户会话类 #

创建一个类封装每个用户的专属数据（语言、游戏状态、进度等），替代全局变量。

# utils/user_session.py
class UserSession:
    def __init__(self, user_id: int):
        self.user_id = user_id
        self.language = "zh-CN"  # 默认语言
        self.game_mode = None  # 存储当前选择的模式：truth/dare
        self.current_round = 0  # 游戏回合数
        # 可根据需求添加更多用户专属属性

# 全局字典存储所有用户会话，key为用户ID
user_sessions = {}

def get_user_session(user_id: int) -> UserSession:
    """获取或初始化用户会话"""
    if user_id not in user_sessions:
        user_sessions[user_id] = UserSession(user_id)
    return user_sessions[user_id]

2. 修改处理器逻辑 #

所有处理器中不再使用全局变量，通过get_user_session获取当前用户的专属会话对象，操作其中的属性。

# handlers/game.py
from aiogram import types
from loader import dp
from utils.user_session import get_user_session

@dp.message_handler(commands=['start'])
async def cmd_start(message: types.Message):
    user_session = get_user_session(message.from_user.id)
    # 初始化用户游戏状态
    user_session.game_mode = None
    user_session.current_round = 0
    await message.answer(f"欢迎来到真心话大冒险！选择模式：/truth 真心话 | /dare 大冒险")

@dp.message_handler(commands=['truth'])
async def cmd_truth(message: types.Message):
    user_session = get_user_session(message.from_user.id)
    user_session.game_mode = "truth"
    # 根据用户语言获取对应问题
    question = get_truth_question(user_session.language)
    await message.answer(question)

@dp.message_handler(commands=['dare'])
async def cmd_dare(message: types.Message):
    user_session = get_user_session(message.from_user.id)
    user_session.game_mode = "dare"
    task = get_dare_task(user_session.language)
    await message.answer(task)

# 本地化内容根据用户会话的language属性返回
def get_truth_question(language: str) -> str:
    questions = {
        "zh-CN": "你最不想让别人知道的秘密是什么？",
        "en-US": "What's a secret you never want anyone to know?"
    }
    return questions.get(language, questions["zh-CN"])

def get_dare_task(language: str) -> str:
    tasks = {
        "zh-CN": "给通讯录第3个人发一句「我喜欢你」",
        "en-US": "Send 'I like you' to the 3rd person in your contacts"
    }
    return tasks.get(language, tasks["zh-CN"])

3. 正确使用FSM状态 #

Aiogram的FSMStorage本身就是按user_id + chat_id隔离的，只要正确使用状态上下文，不会出现多用户混乱：

# handlers/game_states.py
from aiogram.dispatcher.filters.state import State, StatesGroup
from aiogram.dispatcher import FSMContext
from aiogram import types
from loader import dp
from utils.user_session import get_user_session

class GameStates(StatesGroup):
    WaitingForAnswer = State()  # 等待用户回答的状态

@dp.message_handler(state=GameStates.WaitingForAnswer)
async def handle_answer(message: types.Message, state: FSMContext):
    user_session = get_user_session(message.from_user.id)
    # 仅处理当前用户的回答逻辑
    await message.answer(f"你的回答已收到！继续游戏请输入/next")
    await state.finish()  # 结束当前状态

4. 优化内存占用（可选） #

为避免长时间运行导致内存泄漏，可定期清理过期会话：

# utils/user_session.py
import time
from datetime import datetime, timedelta

# 给UserSession添加最后活跃时间属性
class UserSession:
    def __init__(self, user_id: int):
        self.user_id = user_id
        self.language = "zh-CN"
        self.game_mode = None
        self.current_round = 0
        self.last_active = datetime.now()

def cleanup_inactive_sessions():
    """清理30天未活跃的用户会话"""
    cutoff_time = datetime.now() - timedelta(days=30)
    inactive_ids = [uid for uid, session in user_sessions.items() if session.last_active < cutoff_time]
    for uid in inactive_ids:
        del user_sessions[uid]

# 可在启动时添加定时器，每天执行一次清理
# 示例：用threading.Timer实现
import threading
def schedule_cleanup():
    cleanup_inactive_sessions()
    threading.Timer(86400, schedule_cleanup).start()  # 24小时后再次执行

# 在app.py启动时调用
# schedule_cleanup()

关键注意事项 #

• 禁止使用全局变量存储用户专属数据：所有用户相关状态、配置都必须放在UserSession实例中。
• 本地化逻辑必须绑定用户会话：不要使用全局的本地化变量，而是根据用户会话的language属性动态获取内容。
• Aiogram的Bot/Dispatcher单例无需修改：框架本身的单例设计是合理的，无需为每个用户创建独立Bot实例。

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