thread.join()阻塞异步函数：python-telegram-bot股票筛选器无阻塞运行问题

问题根源 #

原代码里的compute.join()会直接阻塞当前的事件循环线程，导致机器人在等待计算完成的时间段内完全无法响应用户交互，彻底违背了异步设计的初衷。另外，在线程中调用asyncio.run()会启动一个独立的新事件循环，和机器人的主事件循环完全隔离，后续发送消息的逻辑也无法和主循环正常配合。

解决方案 #

根据你的需求，推荐以下几种更合理的实现方式：

方式一：用asyncio.create_task直接异步执行任务 #

如果你的heavy_computations本身是异步函数（比如基于aiohttp的异步数据请求、异步IO操作），直接用asyncio的任务调度即可，不需要额外线程：

async def run_screener(bot):
    while True:
        async def heavy_computations():
            results = []
            for i in range(5):
                await asyncio.sleep(2)
                print("Doing computations")
                # 替换为实际的股票数据获取与计算逻辑
                results.append(f"计算结果{i}")
            return results

        # 提交异步任务，不阻塞当前循环
        compute_task = asyncio.create_task(heavy_computations())
        # 异步等待任务完成，期间事件循环可处理其他请求
        results = await compute_task

        # 向订阅用户发送结果
        for user_id in users:
            await bot.send_message(text="\n".join(results), chat_id=user_id)

        await asyncio.sleep(compute_next_time())

方式二：用线程池处理CPU密集型同步任务 #

如果你的计算是CPU密集型的同步代码（比如大量数学运算，无法异步化），可以用asyncio的run_in_executor将任务放到线程池执行，避免阻塞主事件循环：

import concurrent.futures
import time

# 定义同步计算函数（不要用async def）
def heavy_computations_sync():
    results = []
    for i in range(5):
        time.sleep(2)
        print("Doing computations")
        results.append(f"计算结果{i}")
    return results

async def run_screener(bot):
    # 创建线程池，可指定最大线程数
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)
    while True:
        # 将同步任务提交到线程池，异步等待结果
        results = await asyncio.get_event_loop().run_in_executor(executor, heavy_computations_sync)

        # 发送计算结果
        for user_id in users:
            await bot.send_message(text="\n".join(results), chat_id=user_id)

        await asyncio.sleep(compute_next_time())

方式三：修正线程调用逻辑（不推荐，仅作参考） #

如果坚持要用单独线程，绝对不能调用join()，而是通过asyncio的Future对象通知主循环任务完成：

import threading

async def run_screener(bot):
    while True:
        # 创建Future对象，用于接收线程计算结果
        result_future = asyncio.Future()

        def thread_target():
            # 执行同步计算逻辑
            results = []
            for i in range(5):
                time.sleep(2)
                print("Doing computations")
                results.append(f"计算结果{i}")
            # 将结果传入Future，通知主循环任务完成
            asyncio.run_coroutine_threadsafe(result_future.set_result(results), asyncio.get_event_loop())

        # 启动线程，不调用join
        compute_thread = threading.Thread(target=thread_target)
        compute_thread.start()

        # 异步等待结果，期间主循环可处理用户交互
        results = await result_future

        # 发送结果给用户
        for user_id in users:
            await bot.send_message(text="\n".join(results), chat_id=user_id)

        await asyncio.sleep(compute_next_time())

关键注意事项 #

• 禁止在非主事件循环的线程中直接调用bot.send_message这类异步方法，必须通过asyncio.run_coroutine_threadsafe将协程提交到主循环执行。
• IO密集型任务优先用原生异步协程，CPU密集型任务优先用线程池/进程池，以此最大化性能。

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