如何在 Aiogram 中实现双用户配对与消息等待机制

本文介绍如何使用 aiogram 构建一个轻量级双用户配对系统：用户 a 启动会话后获得唯一配对码，用户 b 输入该码即完成绑定，双方随即建立单次通信通道，并支持状态持久化与自动清理。

本文介绍如何使用 aiogram 构建一个轻量级双用户配对系统：用户 a 启动会话后获得唯一配对码，用户 b 输入该码即完成绑定，双方随即建立单次通信通道，并支持状态持久化与自动清理。

在构建互动型 Telegram Bot（如双人问答、配对游戏或协作工具）时，常需让两个独立用户临时“连接”——例如用户 A 发起请求后，系统生成唯一配对码；用户 B 输入该码即触发事件，使 A 即时收到通知。这本质上是一个带状态的跨会话消息等待机制，不能依赖内存变量（易丢失），而应结合数据库实现可靠协调。

以下以 sqlite 为例，展示完整、健壮的实现方案（兼容 Aiogram 3.x）：

✅ 核心设计思路

• 使用一张全局配对表 pairing_requests，存储待匹配的请求；
• 每条记录包含：id（自增主键）、initiator_id（用户 A 的 user.id）、code（随机生成的唯一字符串）、status（’pending’ / ‘matched’）、created_at；
• 不为每对用户动态建表（原文中 CREATE table {user1_id + user2_id} 方案存在 SQL 注入风险、表名非法字符问题且难以维护），而是统一管理、索引优化。

? 示例代码（Aiogram 3 + sqlite）

import sqlite3 import random from aiogram import Router, F from aiogram.types import Message from aiogram.filters import Command  router = Router()  # 初始化数据库（建议在 bot 启动时执行一次） def init_db():     conn = sqlite3.connect("bot.db")     cur = conn.cursor()     cur.execute("""         CREATE TABLE IF NOT EXISTS pairing_requests (             id INTEGER PRIMARY KEY AUTOINCREMENT,             initiator_id INTEGER NOT NULL,             code TEXT UNIQUE NOT NULL,             status TEXT DEFAULT 'pending',             created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP         )     """)     conn.commit()     conn.close()  # 预定义词库（确保无重复、无敏感字符） CODE_WORDS = ["loremipsum", "randomtext", "quantumflux", "nebulaforge", "stardustkey"]  @router.message(Command("start")) async def cmd_start(message: Message):     user_id = message.from_user.id     code = random.choice(CODE_WORDS)      # 存储配对请求     conn = sqlite3.connect("bot.db")     cur = conn.cursor()     cur.execute(         "INSERT INTO pairing_requests (initiator_id, code) VALUES (?, ?)",         (user_id, code)     )     conn.commit()     conn.close()      await message.answer(         f"✅ 已创建配对请求！n请让另一位用户向本机器人发送以下指令：nn`/{code}`nn（注意：斜杠不可省略）",         parse_mode="Markdown"     )  @router.message(F.text.regexp(r"^/([a-zA-Z0-9]+)$")) async def handle_code_command(message: Message):     code = message.text[1:]  # 去掉开头的 '/'      conn = sqlite3.connect("bot.db")     cur = conn.cursor()     cur.execute(         "select initiator_id FROM pairing_requests "         "WHERE code = ? AND status = 'pending'",         (code,)     )     row = cur.fetchone()      if row:         initiator_id = row[0]         # 标记为已匹配（防止重复触发）         cur.execute(             "UPDATE pairing_requests SET status = 'matched' WHERE code = ?",             (code,)         )         conn.commit()          # 通知发起者         await message.bot.send_message(             chat_id=initiator_id,             text=f"? 用户 `{message.from_user.id}` 已成功输入配对码 `{code}`！"         )         await message.answer("✅ 配对成功！对方已收到通知。")     else:         await message.answer("❌ 无效或已使用的配对码，请检查拼写或联系发起者。")      conn.close()

⚠️ 关键注意事项

• 安全性：避免直接拼接用户输入进 SQL（如原文 f”’CREATE TABLE {user1_id + user2_id}…）；始终使用参数化查询。
• 幂等性：通过 status 字段确保同一配对码仅触发一次响应，防止刷屏或重复通知。
• 可扩展性：如需支持多轮配对或超时自动清理，可在表中增加 expires_at 字段，并配合后台任务定期清理过期记录。
• 错误处理：生产环境应补充异常捕获（如数据库连接失败）、日志记录及用户友好提示。
• 并发安全：SQLite 在多数 Bot 场景下足够，若并发极高，建议切换至 postgresql 并加行级锁（SELECT … FOR UPDATE）。

✅ 总结

该方案摒弃了动态建表等高风险做法，采用标准化关系模型，兼顾简洁性、可维护性与可靠性。通过 code 作为轻量级“握手凭证”，实现了跨用户、跨会话的状态同步——这是构建多人协作类 Bot 的基础能力。后续可基于此扩展为邀请链接、临时房间、双人投票等更复杂场景。