如何让异步函数在全局变量更新至指定值后终止？

我要开发一段监听Twitch频道积分奖励兑换的代码，需求是：当特定奖励被兑换时，获取兑换者用户名，然后持续打印该用户的消息直到停止。目前各功能模块都已实现，但run_points_redeem_watcher()异步函数无法在全局变量redeemName匹配目标值时自动终止。当前通过input()手动终止可正常运行，但需要实现自动检测全局变量达标后终止的逻辑，且要正确关闭PubSub与Twitch连接。尝试过atexit、.exit()/.cancel()/.stop()`等方法，要么无效，要么未正确释放资源或直接终止整个程序，均未达到预期效果。

from twitchAPI.pubsub import PubSub
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
from twitchAPI.type import AuthScope, ChatEvent
from twitchAPI.chat import Chat, EventData, ChatMessage
from twitchAPI.oauth import UserAuthenticator
import asyncio
from uuid import UUID

APP_ID = 'App_Id_value'
APP_SECRET = 'App_Secret_value'
USER_SCOPE = [AuthScope.CHANNEL_READ_REDEMPTIONS, AuthScope.CHAT_READ]
TARGET_CHANNEL = 'channel_name_value'
redeemName = ""
redeemerName = ""

async def on_ready(ready_event: EventData):
    print('Bot is ready for work, joining channels')
    await ready_event.chat.join_room(TARGET_CHANNEL)

async def on_message(msg):
    if redeemerName.lower() == msg.user.name:
        print("the selected user said " + msg.text)

async def callback_points(uuid: UUID, data: dict):
    print('got callback for UUID ' + str(uuid))
    print(data["data"]["redemption"]["reward"]["title"])
    global redeemName
    redeemName = data["data"]["redemption"]["reward"]["title"]
    global redeemerName
    redeemerName = data["data"]["redemption"]["user"]["display_name"]


async def run_points_redeem_watcher():
    # setting up Authentication and getting your user id
    twitch = await Twitch(APP_ID, APP_SECRET)
    auth = UserAuthenticator(twitch, USER_SCOPE, force_verify=False)
    token, refresh_token = await auth.authenticate()
    print("auth success")
    await twitch.set_user_authentication(token, USER_SCOPE, refresh_token)
    user = await first(twitch.get_users(logins=[TARGET_CHANNEL]))

    # starting up PubSub
    pubsub = PubSub(twitch)
    pubsub.start()
    uuid = await pubsub.listen_channel_points(user.id, callback_points)

    #The current code looks like this
    input('press ENTER to close...')
    pubsub.stop()
    await twitch.close()
    
    #What I would want to have work is something like this in theory
    if redeemName == "Correct Redeem Name Value":
        pubsub.stop()
        await twitch.close()

async def run_user_focused_chat():
    twitch = await Twitch(APP_ID, APP_SECRET)
    auth = UserAuthenticator(twitch, USER_SCOPE, force_verify=False)
    token, refresh_token = await auth.authenticate()
    print("auth success")
    print("redeemer name is " + redeemerName)

    await twitch.set_user_authentication(token, USER_SCOPE, refresh_token)
    user = await first(twitch.get_users(logins=[TARGET_CHANNEL]))

    chat = await Chat(twitch)
    chat.register_event(ChatEvent.READY, on_ready)
    chat.start()
    chat.register_event(ChatEvent.MESSAGE, on_message)
    input('press ENTER to close...')
    chat.stop()
    await twitch.close()

asyncio.run(run_points_redeem_watcher())
if redeemName == "Correct Redeem Name Value":
    asyncio.run(run_user_focused_chat())

核心问题是全局变量在异步环境中检测不可靠，且两次调用asyncio.run会创建独立事件循环，导致资源管理混乱。改用asyncio.Event实现异步任务间的安全通信，同时将两个流程整合到同一个事件循环中执行：

from twitchAPI.pubsub import PubSub
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
from twitchAPI.type import AuthScope, ChatEvent
from twitchAPI.chat import Chat, EventData, ChatMessage
from twitchAPI.oauth import UserAuthenticator
import asyncio
from uuid import UUID

APP_ID = 'App_Id_value'
APP_SECRET = 'App_Secret_value'
USER_SCOPE = [AuthScope.CHANNEL_READ_REDEMPTIONS, AuthScope.CHAT_READ]
TARGET_CHANNEL = 'channel_name_value'
TARGET_REDEEM_NAME = "Correct Redeem Name Value"  # 替换为你的目标奖励名称
redeemerName = ""
# 用异步事件替代全局变量，实现安全的任务通知
redeem_triggered = asyncio.Event()

async def on_ready(ready_event: EventData):
    print('Bot is ready for work, joining channels')
    await ready_event.chat.join_room(TARGET_CHANNEL)

async def on_message(msg):
    if redeemerName.lower() == msg.user.name:
        print(f"[{msg.user.name}] {msg.text}")

async def callback_points(uuid: UUID, data: dict):
    global redeemerName
    current_redeem = data["data"]["redemption"]["reward"]["title"]
    print(f'收到积分兑换回调（UUID: {uuid}）：{current_redeem}')
    
    # 匹配到目标奖励时，记录用户名并触发事件
    if current_redeem == TARGET_REDEEM_NAME:
        redeemerName = data["data"]["redemption"]["user"]["display_name"]
        print(f'触发目标奖励！兑换用户：{redeemerName}')
        redeem_triggered.set()  # 通知监听任务可以终止

async def run_points_redeem_watcher():
    twitch = await Twitch(APP_ID, APP_SECRET)
    auth = UserAuthenticator(twitch, USER_SCOPE, force_verify=False)
    token, refresh_token = await auth.authenticate()
    print("认证成功")
    await twitch.set_user_authentication(token, USER_SCOPE, refresh_token)
    user = await first(twitch.get_users(logins=[TARGET_CHANNEL]))

    # 启动PubSub监听
    pubsub = PubSub(twitch)
    pubsub.start()
    await pubsub.listen_channel_points(user.id, callback_points)
    
    print(f"开始监听{TARGET_CHANNEL}的积分奖励兑换...")
    # 等待目标奖励触发事件
    await redeem_triggered.wait()
    
    # 事件触发后，正确关闭资源
    pubsub.stop()
    await twitch.close()
    print("PubSub连接已关闭")

async def run_user_focused_chat():
    print(f"开始监听用户{redeemerName}的消息...")
    twitch = await Twitch(APP_ID, APP_SECRET)
    auth = UserAuthenticator(twitch, USER_SCOPE, force_verify=False)
    token, refresh_token = await auth.authenticate()
    await twitch.set_user_authentication(token, USER_SCOPE, refresh_token)
    user = await first(twitch.get_users(logins=[TARGET_CHANNEL]))

    chat = await Chat(twitch)
    chat.register_event(ChatEvent.READY, on_ready)
    chat.register_event(ChatEvent.MESSAGE, on_message)
    chat.start()
    
    # 保持聊天监听，直到用户手动终止
    await asyncio.get_event_loop().run_in_executor(None, input, "按ENTER停止监听消息...")
    
    chat.stop()
    await twitch.close()
    print("聊天监听已关闭")

async def main():
    # 先运行积分奖励监听，直到目标奖励被触发
    await run_points_redeem_watcher()
    # 触发后运行聊天监听
    await run_user_focused_chat()

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

关键修改说明 #

1. 用asyncio.Event替代全局变量：异步事件是线程/任务安全的通信方式，避免全局变量在异步环境中的检测延迟问题
2. 整合事件循环：将两个异步任务放到同一个main函数中，通过asyncio.run一次执行，避免多次创建事件循环导致的资源冲突
3. 正确的资源关闭流程：在run_points_redeem_watcher中等待事件触发后，先停止PubSub再关闭Twitch连接，确保资源被正确释放
4. 明确目标奖励常量：将目标奖励名称定义为常量TARGET_REDEEM_NAME，便于维护

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