无需事件循环库,在游戏Tick中处理Python协程并实现非阻塞async sleep函数
无需事件循环库,在游戏Tick中处理Python协程并实现非阻塞async sleep函数
我完全懂你的需求——要一个不阻塞、适配游戏Tick机制的async sleep函数,而且不能用额外的库对吧?之前你尝试用yield导致报错,是因为async函数里直接yield会变成异步生成器,而await需要的是标准的可等待对象(coroutine、Future等),咱们来一步步解决这个问题。
核心思路是:
- 维护一个任务列表,用来跟踪所有需要在Tick中推进的协程
- 在每个Tick里,逐个推进这些协程,直到它们完成
- sleep函数将总时长拆分成多个Tick间隔,通过多次让出控制权来模拟“睡眠”
下面是修改后的完整代码:
import time # do not import any other libs, complete my functions # 全局任务列表:存储需要在Tick中推进的协程生成器 tasks = [] def tick(): global tasks completed_tasks = [] # 遍历所有任务,逐个推进协程 for idx, coro_gen in enumerate(tasks): try: # 推进协程的执行,next()对应一次yield/await的切换 next(coro_gen) except StopIteration: # 协程执行完成,标记为待移除 completed_tasks.append(idx) # 从后往前移除完成的任务,避免索引错乱 for idx in reversed(completed_tasks): del tasks[idx] def run_coroutine(coro): """将async协程转换为可推进的生成器,加入任务列表""" coro_generator = coro.__await__() tasks.append(coro_generator) async def sleep(seconds): # 每个Tick的间隔(和下方模拟循环的time.sleep(0.01)保持一致) tick_interval = 0.01 # 计算需要等待的总Tick数 total_ticks = int(seconds / tick_interval) # 定义一个极简的可等待对象,用来让出控制权到下一个Tick class TickAwaitable: def __await__(self): yield # 逐个Tick等待,直到完成总时长 for _ in range(total_ticks): await TickAwaitable() async def test(): print("start") await sleep(2) print("end") # 把test协程加入任务列表,等待Tick推进 run_coroutine(test()) # simulate game tick while True: tick() time.sleep(0.01)
代码说明:
任务列表与tick函数:
tasks列表存储所有需要处理的协程生成器(由async函数的__await__()方法生成)tick()函数在每个游戏Tick中遍历任务列表,调用next()推进协程;如果协程抛出StopIteration,说明它已经执行完成,就从列表中移除
run_coroutine函数:
- 把async协程转换成可迭代的生成器,加入任务列表,这样tick函数才能推进它的执行
sleep函数:
- 根据总时长和Tick间隔计算需要等待的Tick数量
- 定义了一个极简的
TickAwaitable类,它的__await__()方法会yield一次,让出控制权到下一个Tick - 通过循环await这个对象,实现分Tick的非阻塞等待
这样运行代码后,你会看到先打印"start",等待2秒(200个Tick)后打印"end",整个过程完全基于游戏Tick机制,没有阻塞,也不需要依赖任何额外的事件循环库。
备注:内容来源于stack exchange,提问作者xiaoguoNo1
相关产品推荐
相关产品推荐

