线程无法获取异步函数返回值，报TypeError: 'coroutine'对象不可下标

错误原因 #

mqtt.publish_message是基于asyncio的异步函数，直接调用它不会执行函数逻辑，而是返回一个协程对象（coroutine object）。你的代码中将这个协程对象直接赋值给self.value，后续尝试用下标results['request']访问时自然会触发TypeError。

同时，原代码的线程使用逻辑存在两个问题：

1. 线程仅能启动一次，原代码未正确处理线程重复执行的场景
2. 未在事件循环中运行异步函数，无法获取实际返回值

解决方案 #

方案一：单次任务线程（适合不频繁的单次异步调用） #

修改线程类，在run方法中使用asyncio.run运行异步函数并获取返回值，同时调整主线程的线程创建逻辑：

from threading import Thread
import asyncio

class EncodingThread(Thread):
    def __init__(self):
        super().__init__()
        self.value = None
        self.frame = None

    def setFrame(self, frame):
        self.frame = frame

    def run(self):
        # 在事件循环中运行异步函数，获取实际返回值
        self.value = asyncio.run(mqtt.publish_message(self.frame))

def somedef():    
    while True:
        # 此处执行你的其他业务逻辑，不受异步任务干扰
        
        # 每次需要执行异步任务时，创建新线程并启动
        thread = EncodingThread()
        thread.setFrame(some_value)
        thread.start()

        # 检查线程状态，完成后获取结果
        thread_status = thread.is_alive()
        if not thread_status:
            thread.join()
            results = thread.value
            # 修正原逻辑的判断错误：(1 and None)等价于None，此处改为判断结果不为1和None
            if results not in (1, None):
                value = results['request']
                return value
        
        # 继续执行其他业务逻辑...

方案二：常驻任务线程（适合频繁的异步调用，减少线程创建开销） #

让线程持续运行，通过队列接收任务，用事件标记任务完成，避免重复创建线程：

from threading import Thread, Event
import asyncio
import queue

class EncodingThread(Thread):
    def __init__(self):
        super().__init__()
        self.value = None
        self.frame_queue = queue.Queue()
        self.running = True
        self.task_done = Event()  # 标记异步任务完成

    def setFrame(self, frame):
        self.value = None  # 重置结果
        self.task_done.clear()  # 清除完成标记
        self.frame_queue.put(frame)

    def run(self):
        # 为子线程创建独立的事件循环
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        
        while self.running:
            try:
                # 等待队列中的任务，超时1秒避免阻塞线程退出
                frame = self.frame_queue.get(timeout=1)
                # 运行异步函数并保存结果
                self.value = loop.run_until_complete(mqtt.publish_message(frame))
                self.task_done.set()  # 标记任务完成
                self.frame_queue.task_done()
            except queue.Empty:
                continue
        
        loop.close()

    def stop(self):
        # 停止线程
        self.running = False
        self.join()

# 初始化并启动常驻线程
thread = EncodingThread()
thread.start()

def somedef():    
    while True:
        # 此处执行你的其他业务逻辑
        
        # 提交异步任务
        thread.setFrame(some_value)

        # 检查任务是否完成，完成后获取结果
        if thread.task_done.is_set():
            results = thread.value
            if results not in (1, None):
                value = results['request']
                return value
        
        # 继续执行其他业务逻辑...

# 程序退出时记得停止线程
# thread.stop()

关键改动说明 #

1. 异步函数执行：通过asyncio.run()或loop.run_until_complete()在事件循环中运行异步函数，获取实际返回值而非协程对象
2. 线程逻辑优化：
   • 方案一每次创建新线程，适合单次任务
   • 方案二使用队列和事件实现常驻线程，适合频繁调用场景，减少线程创建销毁的开销
3. 条件判断修正：原代码results != (1 and None)逻辑错误，(1 and None)等价于None，改为results not in (1, None)更符合需求

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