Python如何将async异步函数作为同步回调参数传入类构造函数

错误根因说明 #

你遇到的两个报错原因非常明确：

• 直接传入异步函数的类型错误：异步函数调用后返回的是协程对象，和回调要求的(bytes) -> None类型的返回值不匹配。
• asyncio.run触发的运行时错误：asyncio.run是用来启动全新事件循环的接口，你的代码外层已经通过asyncio.run(main())启动了一个运行中的事件循环，异步框架不允许嵌套启动事件循环。

通用解决方案 #

适配逻辑核心是：在同步回调包装函数中，将异步函数的协程提交到当前已经运行的事件循环中调度执行，不需要启动新循环，同时满足返回None的类型要求。

场景1：回调在事件循环所在线程触发 #

如果data_receiver只会在事件循环运行的主线程被调用，直接用asyncio.create_task提交协程即可：

import asyncio
from typing import Callable

class SomeClass:
    data_receiver: Callable[[bytes], None]
    def __init__(self, data_receiver: Callable[[bytes], None]):
        self.data_receiver = data_receiver

async def some_async_function(data: bytes):
    # 你的异步业务逻辑
    print(f"处理数据: {data}")
    await asyncio.sleep(1)
    print("异步逻辑执行完成")

async def main():
    # 同步包装函数，完全匹配回调类型要求
    def sync_data_receiver(data: bytes) -> None:
        # 提交协程到当前事件循环调度，无阻塞，返回None
        asyncio.create_task(some_async_function(data))
    
    obj = SomeClass(sync_data_receiver)
    obj.data_receiver(b"test payload")

    # 若为短期运行脚本，加这段等待所有异步任务执行完成；长期运行服务可删除
    pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    await asyncio.gather(*pending)

asyncio.run(main())

场景2：回调在其他子线程触发 #

如果data_receiver会在非事件循环所在的其他线程被调用，需要使用线程安全的提交接口：

async def main():
    # 提前获取当前运行的事件循环实例
    loop = asyncio.get_running_loop()

    def sync_data_receiver(data: bytes) -> None:
        # 跨线程提交任务，线程安全
        loop.call_soon_threadsafe(asyncio.create_task, some_async_function(data))
    
    obj = SomeClass(sync_data_receiver)
    # 可模拟其他线程调用 obj.data_receiver(...)
    obj.data_receiver(b"cross thread test")

    pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    await asyncio.gather(*pending)

asyncio.run(main())

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