如何将Python异步函数作为C回调函数使用（基于ctypes）

我需要调用一个C库，让它调用Python函数处理IO操作，以此避免编写对应的C IO实现。现有Python代码大量使用asyncio，所有IO操作均通过队列完成，且通过ctypes加载该C库。但问题在于无法在C回调中使用await语法，请问是否有办法将Python异步函数作为C回调函数使用？

除了示例中类似non_async_callback的同步轮询方案外，是否存在通过ctypes在C回调中实现await的方式？

Python 示例代码 #

import asyncio
import time
import ctypes

DLL_PATH="./bin/test.dll"

dll = ctypes.CDLL(DLL_PATH)
incoming_queue=asyncio.Queue()
outgoing_queue=asyncio.Queue()

c_func =  ctypes.CFUNCTYPE(ctypes.c_uint8, ctypes.c_uint8)

async def python_callback(test_num):
    await outgoing_queue.put(test_num)
    test_num = await incoming_queue.get()
    return test_num

def non_async_callback(test_num):
    while True:
        try:
            outgoing_queue.put_nowait(test_num)
            break
        except asyncio.QueueFull:
            time.sleep(0.1)
            continue
   
    while True:
        try:
            test_num = incoming_queue.get_nowait()
            break 
        except asyncio.QueueEmpty:
            time.sleep(0.1)
            continue

    return test_num

# 初始化阶段调用
def setup_callback():
    dll.setup_callback(c_func(python_callback))

C 示例代码 #

#include <stdint.h>

typedef uint8_t (*CALLBACK)(uint8_t);
CALLBACK py_callback;

void setup_callback(void* callback)
{
    py_callback = (CALLBACK)callback;
}

// 由C代码其他逻辑触发调用
uint8_t python_callback(uint8_t value)
{
    uint8_t result = py_callback(value);
    // 处理返回结果
    return result;
}

直接在C回调中使用await是不可能的，因为C回调是同步阻塞调用，而await属于Python异步语法，只能在async函数内使用，且依赖asyncio的事件循环调度。不过可以通过以下方式实现异步逻辑的适配，避免低效轮询：

方案1：用事件循环同步接口调度异步任务 #

在同步回调中，获取当前运行的asyncio事件循环，通过run_until_complete()直接运行异步回调函数，等待其完成后返回结果：

def callback_adapter(test_num):
    loop = asyncio.get_running_loop()
    # 阻塞等待异步任务执行完成
    result = loop.run_until_complete(python_callback(test_num))
    return result

# 注册适配器作为C回调
def setup_callback():
    dll.setup_callback(c_func(callback_adapter))

注意：如果C回调是在非事件循环线程中被调用，需要改用线程安全的方式调度任务并同步获取结果：

def thread_safe_callback_adapter(test_num):
    loop = asyncio.get_running_loop()
    # 线程安全地提交异步任务到事件循环
    future = asyncio.run_coroutine_threadsafe(python_callback(test_num), loop)
    # 阻塞等待任务完成并获取结果
    return future.result()

方案2：异步触发式回调（若C库支持） #

如果C库支持异步回调模式（即调用回调后立即返回，不等待结果，待IO完成后再通知Python），可以将异步逻辑完全交由asyncio调度，C代码仅负责触发事件而非等待结果。但如果C库必须同步等待返回值，方案1更直接。

方案3：用asyncio.to_thread适配（Python 3.9+） #

将同步回调中的逻辑放到独立线程执行，避免阻塞主事件循环：

async def async_handler(test_num):
    return await python_callback(test_num)

def callback_adapter(test_num):
    result = asyncio.run(asyncio.to_thread(async_handler, test_num))
    return result

• 线程安全：若C代码在独立线程中调用回调，必须确保asyncio事件循环已启动，且使用线程安全的接口调度异步任务。
• 性能对比：轮询方案会浪费CPU资源且响应延迟高，通过事件循环调度异步任务的方式能更高效地利用系统资源。

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