如何非阻塞运行TwitchIO Bot 无需run()方法占用程序主循环

解决方案 #

TwitchIO 基于异步 asyncio 框架设计，本身没有提供类似 poll() 的同步轮询方法，你可以通过以下两种方式实现非阻塞运行：

方案1：你的主程序本身是异步（asyncio）架构 #

bot.run() 本质是对 asyncio.run(bot.start()) 的封装，会直接占用当前线程的事件循环。你可以手动将 bot 启动任务加入你自己的事件循环，和其他异步任务并行运行：

import asyncio
from twitchio.ext import commands

class Bot(commands.Bot):
    def __init__(self):
        super().__init__(token='ACCESS_TOKEN', prefix='?', initial_channels=['你的频道名'])

    async def event_ready(self):
        print(f'Logged in as | {self.nick}')

    @commands.command()
    async def hello(self, ctx: commands.Context):
        await ctx.send(f'Hello {ctx.author.name}!')

# 你自己的其他异步主逻辑
async def your_other_main_task():
    while True:
        # 这里写你自己的业务逻辑
        print("主程序其他逻辑运行中")
        await asyncio.sleep(1)

async def main():
    bot = Bot()
    # 同时启动bot和你自己的任务
    async with bot:
        await asyncio.gather(
            bot.start(),
            your_other_main_task()
        )

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

方案2：你的主程序是同步架构 #

如果你的主程序是同步写法，没法直接适配 asyncio，直接把 TwitchIO bot 放在单独的守护线程中运行即可，主线程的逻辑不会被阻塞：

import threading
from twitchio.ext import commands

class Bot(commands.Bot):
    def __init__(self):
        super().__init__(token='ACCESS_TOKEN', prefix='?', initial_channels=['你的频道名'])

    async def event_ready(self):
        print(f'Logged in as | {self.nick}')

    @commands.command()
    async def hello(self, ctx: commands.Context):
        await ctx.send(f'Hello {ctx.author.name}!')

def run_bot():
    bot = Bot()
    bot.run()

if __name__ == "__main__":
    # 启动bot线程，设置为守护线程，主程序退出时会自动关闭bot
    bot_thread = threading.Thread(target=run_bot, daemon=True)
    bot_thread.start()

    # 这里写你自己的同步主循环逻辑，不会被阻塞
    while True:
        print("主程序正常运行中")
        # 你的其他业务逻辑
        input()

注意事项 #

• 如果使用多线程方案，跨线程调用 bot 的发送消息等异步方法时，需要用 asyncio.run_coroutine_threadsafe 来提交任务到 bot 所在的事件循环，避免线程安全问题。
• bot 的事件处理逻辑（如命令响应、消息接收）本身就是异步回调触发的，不需要你手动轮询处理消息。

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