在Python异步编程的实际开发中，我们经常会遇到这样的需求：同一个异步函数，既希望它能在其他异步函数中通过await关键字等待执行完成并获取结果，也希望能在同步上下文或者不需要等待结果的场景中直接调用触发执行，不需要阻塞当前流程。要实现这个效果，需要理解Python异步函数的执行机制，再结合对应的封装方法。
异步函数的基础执行逻辑
Python中使用async def定义的函数是异步函数，直接调用异步函数并不会立即执行函数内部的逻辑，而是会返回一个协程对象。如果直接调用异步函数而不做处理，函数内部的代码不会执行，也不会产生任何效果。
要让异步函数真正执行，有两种常见方式：一种是在异步函数中通过await 协程对象的方式等待执行，另一种是通过事件循环来运行协程对象。我们可以通过一个简单的示例来验证这个逻辑：
import asyncio
async def demo_async_func():
print("异步函数开始执行")
await asyncio.sleep(1)
print("异步函数执行完成")
return "执行结果"
# 直接调用异步函数，只会返回协程对象，不会执行内部逻辑
coroutine = demo_async_func()
print(type(coroutine)) # 输出 <class 'coroutine'>
实现异步函数既能await也能直接调用的方法
方法一：封装为自动触发执行的函数
我们可以封装一个外层函数，当直接调用时自动将协程提交到事件循环执行，当被await时则正常返回协程对象。这种方式适合不需要等待执行结果的场景，直接调用后会立即触发执行，不会阻塞当前代码。
import asyncio
from functools import wraps
def async_compatible(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 获取当前事件循环，如果没有则创建新的
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 创建协程对象
coroutine = func(*args, **kwargs)
# 如果是在异步上下文中调用，直接返回协程对象供await使用
# 这里通过判断调用栈是否在异步函数中实现，简化逻辑：直接提交到事件循环执行
# 注意：这种方式下直接调用会立即执行，await时也能正常等待
# 实际使用时可以根据需求调整判断逻辑
asyncio.ensure_future(coroutine)
return coroutine
return wrapper
@async_compatible
async def task_func():
print("任务开始执行")
await asyncio.sleep(1)
print("任务执行完成")
return "任务结果"
# 直接调用，会立即触发执行，不会阻塞
print("直接调用开始")
task_func()
print("直接调用结束")
# 在异步函数中await调用
async def main():
print("await调用开始")
result = await task_func()
print(f"await调用结果：{result}")
# 运行异步主函数
asyncio.run(main())
方法二：返回可调用对象兼容两种调用方式
另一种思路是让异步函数返回一个可调用对象，当直接调用这个返回的对象时，自动触发协程执行；当被await时，直接等待协程执行完成。这种方式可以更灵活地控制两种调用的行为。
import asyncio
class AsyncCompatibleWrapper:
def __init__(self, coroutine):
self.coroutine = coroutine
def __call__(self):
# 直接调用时，将协程提交到事件循环执行
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
asyncio.ensure_future(self.coroutine)
return self.coroutine
def __await__(self):
# 支持await语法，等待协程执行完成
return self.coroutine.__await__()
def make_async_compatible(func):
@wraps(func)
def wrapper(*args, **kwargs):
coroutine = func(*args, **kwargs)
return AsyncCompatibleWrapper(coroutine)
return wrapper
@make_async_compatible
async def test_func():
print("测试函数执行")
await asyncio.sleep(0.5)
return "测试返回"
# 直接调用触发执行
print("直接调用测试")
test_func()
print("直接调用测试结束")
# await调用等待结果
async def test_main():
print("await调用测试")
res = await test_func()
print(f"await结果：{res}")
asyncio.run(test_main())
两种方式的适用场景对比
我们可以通过表格对比两种实现方式的特点，方便根据实际需求选择：
| 实现方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 自动触发执行封装 | 实现简单，调用方式无感知 | 直接调用时无法立即获取执行结果，依赖事件循环状态 | 不需要等待执行结果，希望直接调用后立即触发执行的场景 |
| 可调用对象包装 | 两种调用方式的行为更可控，支持await获取结果 | 实现相对复杂，需要自定义包装类 | 既需要直接调用触发执行，也需要在异步上下文中等待结果的场景 |
注意事项
- 直接调用异步函数触发执行时，需要确保事件循环处于运行状态，否则协程可能无法正常执行。
- 如果直接调用后需要获取执行结果，不建议使用自动提交到事件循环的方式，因为这种方式下结果需要通过回调或者future对象获取，不如直接使用await方便。
- 在同步代码中直接调用异步函数时，要避免阻塞事件循环，尽量不要在同步代码中用
loop.run_until_complete来运行协程，否则可能会导致事件循环冲突。
通过上述方法，我们可以让同一个异步函数同时支持await调用和直接调用两种使用方式，根据实际场景选择合适的实现方案，能够提升代码的灵活性和复用性。