discord.py开发答题机器人如何创建独立stop命令终止运行中的会话循环

核心思路是用全局状态存储跨命令共享答题会话的运行状态，同时支持异步场景下的终止信号传递，具体修改如下：

1. 定义全局会话存储 #

在所有命令外层添加全局变量，用于记录活跃的答题会话，支持多用户/多频道同时答题互不干扰：

import asyncio
import random

# 存储活跃答题会话：键为 (发起用户ID, 频道ID)，值为终止信号事件
active_quizzes = {}

2. 修改ask命令逻辑 #

添加会话注册、终止信号判断逻辑，替换原来的内部running函数：

@client.command()
async def ask(ctx):
    session_key = (ctx.author.id, ctx.channel.id)
    # 避免同一个用户在同一个频道重复发起答题
    if session_key in active_quizzes:
        await ctx.send("你已经有一个正在进行的答题会话了哦")
        return
    
    # 注册当前会话，创建终止事件
    stop_event = asyncio.Event()
    active_quizzes[session_key] = stop_event

    quiz_data = {
        'question_1': (["1", "one"], "one"),
        'question_2': (["2", "two"], "two"),
    }
    question = random.choice(list(quiz_data.keys()))
    answers, hint = quiz_data[question]

    await ctx.send("What is the answer to this question?")
    await asyncio.sleep(1)
    await ctx.send(question)

    def check_msg(msg):
        # 消息校验同时判断是否触发了终止信号
        return (msg.author == ctx.author 
                and msg.channel == ctx.channel
                and not stop_event.is_set())

    def check_answer(msg):
        return any(answer in msg.content.lower() for answer in answers)
    
    try:
        async with asyncio.timeout(10):
            while not stop_event.is_set():
                msg = await client.wait_for('message', check=check_msg)
                if check_answer(msg):
                    await ctx.send("good")
                    break
                else:
                    print('someone got it wrong lol')
    except asyncio.TimeoutError:
        # 超时后先判断是否已经被手动终止
        if stop_event.is_set():
            del active_quizzes[session_key]
            return
        await ctx.send(f"What is the answer to this question? hint: {hint}")
        await ctx.send(question)

        try:
            msg = await client.wait_for('message', check=check_msg, timeout=10)
            if check_answer(msg):
                await ctx.send("good")
            else:
                await ctx.send("wrong.")
        except asyncio.TimeoutError:
            await ctx.send("答题超时，本次会话结束")
    finally:
        # 所有场景结束后清理会话记录
        if session_key in active_quizzes:
            del active_quizzes[session_key]

3. 重写stopg命令逻辑 #

现在可以直接从全局存储中读取会话状态，触发终止信号：

@client.command()
async def stopg(ctx):
    session_key = (ctx.author.id, ctx.channel.id)
    if session_key in active_quizzes:
        # 触发终止信号
        active_quizzes[session_key].set()
        await ctx.send("Round is stopping...")
    else:
        await ctx.send("No round active")

方案说明 #

• 用asyncio.Event作为终止信号，适配异步场景下的等待逻辑，不需要反复轮询状态
• 基于用户ID+频道ID做会话隔离，支持多用户同时答题互不影响
• finally块保证无论会话是正常结束、超时还是手动终止，都会清理状态，不会出现残留的无效会话
• 如果你的机器人只需要单频道单场答题，也可以简化为单个全局布尔变量作为运行状态标识，逻辑更简单

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