asyncio 协程
结论
KBEngine 支持 Python asyncio,适合处理 HTTP、定时等待和其他能够真正异步挂起的 IO 操作。但 asyncio 不会让 Python 逻辑脱离组件主线程,也不会自动把耗时函数变成非阻塞函数。
判断一个协程是否影响 Tick,不能只看代码中有没有 await,而要看:
| 判断点 | 不明显阻塞 Tick | 会阻塞或延迟 Tick |
|---|---|---|
| 等待对象 | 非阻塞 Socket、asyncio.sleep()、未完成的 Future | 内部仍调用同步 IO 的“伪异步”函数 |
| 协程执行 | 两次 await 之间只有少量 Python 代码 | CPU 密集循环、大量 JSON 解析、批量遍历 Entity |
| 系统调用 | 异步 HTTP 客户端、异步数据库客户端 | time.sleep()、同步 HTTP、同步文件或数据库操作 |
| 任务唤醒 | 少量任务分散恢复 | 大量任务在同一时刻恢复并继续执行 |
准确地说:
非阻塞
await的等待阶段不会占用组件主线程;协程首次执行和每次恢复后的 Python 代码仍在组件主线程执行,因此仍受 Tick 预算约束。
启用 asyncio
asyncio 默认关闭。需要在项目的 res/server/kbengine.xml 中设置 asyncioRepeatOffset:
<root>
<!-- 组件 Dispatcher 推进 asyncio 事件循环的间隔,单位秒。 -->
<!-- Interval in seconds for the component Dispatcher to pump asyncio. -->
<asyncioRepeatOffset>0.01</asyncioRepeatOffset>
</root>| 配置值 | 行为 | 建议 |
|---|---|---|
0 | 关闭引擎托管的 asyncio;返回的协程不会被创建为 Task,并输出配置错误。 | 项目没有使用协程时保持默认。 |
0.01 | 大约每 10 ms 推进一次 asyncio,是常用的压测起始值。 | 对协程恢复延迟有要求时优先从该值开始。 |
0.02 | 大约每 20 ms 推进一次,调度次数更少。 | IO 回调频率较低、希望减少调度开销时测试。 |
小于 0.01 的正值 | 当前配置加载时按 0.01 秒处理。 | 不要依赖更小的配置值获取更高精度。 |
asyncioRepeatOffset 决定 asyncio 事件循环的推进间隔,不等于 gameUpdateHertz,也不会提高游戏逻辑 Tick。间隔越短,协程通常恢复得越及时,但 Dispatcher 唤醒、Python 调度和 Task 检查次数也越多。
引擎执行模型
每个支持 asyncio 的组件拥有自己的事件循环。事件循环由该组件的 Dispatcher 定时推进,不额外创建 Python 调度线程。
组件 Dispatcher 主循环
├── 网络事件
├── 游戏 Tick
├── Timer
└── asyncio 定时推进
├── 执行 ready callback
├── 恢复可运行的 Task
└── 收集完成结果与异常| 当前机制 | 作用 | 边界 |
|---|---|---|
| 独立事件循环 | 每个组件维护自己的 asyncio loop,生命周期跟随组件。 | 不同组件间不能直接共享 Task、Future 或事件循环对象。 |
| 主线程串行执行 | Entity 状态和脚本回调继续遵守组件原有串行模型。 | 协程执行 Python 代码时会占用当前组件主线程。 |
| 非阻塞 pump | 引擎只处理当前已就绪任务,不在事件循环中等待 IO。 | 已经开始执行的 Python 回调不能被 C++ 中途抢占。 |
| 调度预算 | 当前单次推进至少尝试 8 轮、最多 64 轮,并在达到最小轮数后检查约 2 ms 的软时间预算。 | 预算不是硬中断。单个回调执行 50 ms,Tick 仍可能被延迟约 50 ms。 |
| Task 异常回收 | 引擎持有由系统回调返回的 Task,并在完成后读取结果,使异常进入 KBE 日志。 | 脚本自行 create_task() 的后台任务仍应由项目管理。 |
因此,asyncio 能消除的是“等待 IO 时占住主线程”,不能消除 Python 代码本身的 CPU、内存分配、GC、序列化和同步 IO 成本。
await 什么时候会让出执行权
真正的非阻塞等待
import asyncio
async def load_remote_state():
# 等待期间 Task 挂起,组件主线程可以继续处理其他事件。
# The Task is suspended while waiting, allowing the component thread to process other events.
await asyncio.sleep(0.5)
# 协程恢复后,这段 Python 代码仍运行在组件主线程。
# After resuming, this Python code still runs on the component main thread.
return {"ready": True}写了 await 仍可能阻塞
import asyncio
import time
async def blocking_example():
# 同步 sleep 会直接阻塞组件主线程。
# A synchronous sleep directly blocks the component main thread.
time.sleep(1)
# 纯 Python 大循环发生在 await 之前,仍然占用主线程。
# This Python loop runs before await and still monopolizes the main thread.
total = sum(i * i for i in range(5_000_000))
await asyncio.sleep(0.1)
return total异步函数名称也不能证明其内部是非阻塞的。第三方库只有在底层使用非阻塞 Socket、事件循环兼容的 Future,或明确把阻塞工作转移到线程时,等待阶段才会真正让出执行权。
已完成的 awaitable 可能立即继续
如果 Future 已经完成,await future 可能立即取得结果并继续向下执行,不一定把控制权交回组件 Dispatcher。大量“立即完成”的 await 连在一起,仍然可能形成一段长时间的连续执行。
以下写法也不适合作为永久循环:
import asyncio
async def busy_loop():
while True:
# sleep(0) 会让出当前 Task,但它很快再次进入 ready 队列,持续制造调度压力。
# sleep(0) yields this Task, but it quickly re-enters the ready queue and creates scheduling pressure.
await asyncio.sleep(0)周期任务应使用符合业务精度的实际间隔,并在关闭时取消。
适合与不适合的工作
| 工作类型 | 是否推荐直接放在 asyncio 协程中 | 原因与处理方式 |
|---|---|---|
| 异步 HTTP、异步 Socket | 推荐 | 等待网络期间可以挂起,不占用组件主线程。 |
asyncio.sleep() 和超时控制 | 推荐 | 由事件循环调度,不等同于同步休眠。 |
| 少量状态校验和结果应用 | 推荐 | 在组件主线程执行,能够安全地访问当前组件 Entity。 |
| 同步文件、HTTP、数据库调用 | 不推荐 | 会阻塞主线程;改用异步库、专用服务或受控线程。 |
| 大量 JSON、压缩、加密、寻路计算 | 不推荐 | CPU 和内存分配仍发生在主线程,应拆批或移出逻辑进程。 |
| 遍历大量 Entity 并逐个处理 | 不推荐一次完成 | 会造成 Tick 长尾,应限量、分页或跨 Tick 分批。 |
| 无限创建后台 Task | 禁止 | 会增加 Task、Future、闭包和结果对象内存,并形成唤醒风暴。 |
超时与并发限制
外部服务可能永久不返回,生产代码必须设置超时并限制并发量。
import asyncio
from aiohttp import ClientSession
_request_limit = asyncio.Semaphore(32)
async def fetch_json(session: ClientSession, url: str) -> dict:
# 并发上限保护连接数、内存以及同一时刻恢复的 Task 数量。
# The concurrency limit protects connection count, memory, and simultaneous Task wakeups.
async with _request_limit:
# Python 3.12 的 timeout 防止外部服务永久占用任务资源。
# Python 3.12 timeout prevents an external service from retaining Task resources forever.
async with asyncio.timeout(3.0):
async with session.get(url) as response:
response.raise_for_status()
return await response.json()ClientSession 应在组件初始化时创建并复用,在关闭阶段统一执行 await session.close()。不要为每个请求重复创建连接池,否则会增加 Socket、DNS、TLS 握手和临时对象开销。
并发限制需要依据以下指标压测:
| 指标 | 并发过高时的表现 |
|---|---|
| Tick p95/p99 | 多个任务同时恢复,Python 回调时间集中在一个调度周期。 |
| RSS 与 Python 对象数 | Task、响应体、异常和闭包长期占用内存。 |
| 外部连接数 | HTTP 或数据库连接池耗尽。 |
| 网络和 IO | 同一时刻产生大量请求或解析大量响应。 |
| GC | 大批临时对象同时释放,造成周期性停顿。 |
后台任务生命周期
系统回调直接返回的协程会由引擎创建并持有 Task。脚本内部调用 asyncio.create_task() 创建的后台任务,应由项目保留强引用、读取异常并在关闭时取消。
import asyncio
import traceback
from KBEDebug import ERROR_MSG
_background_tasks: set[asyncio.Task] = set()
def _on_task_done(task: asyncio.Task) -> None:
_background_tasks.discard(task)
if task.cancelled():
return
try:
task.result()
except Exception:
# 主动读取并记录异常,避免后台 Task 静默失败。
# Retrieve and log the exception so a background Task cannot fail silently.
ERROR_MSG(traceback.format_exc())
def spawn(coroutine) -> asyncio.Task:
task = asyncio.create_task(coroutine)
_background_tasks.add(task)
task.add_done_callback(_on_task_done)
return task
async def cancel_background_tasks() -> None:
tasks = list(_background_tasks)
for task in tasks:
task.cancel()
if tasks:
# 让 finally 块和取消清理代码获得执行机会。
# Give finally blocks and cancellation cleanup code a chance to run.
await asyncio.gather(*tasks, return_exceptions=True)组件关闭时,引擎会拒绝新任务、取消未完成任务,并有限度地推进事件循环,让 CancelledError 和 finally 有机会执行。但关服清理不能依赖长时间网络请求;重要数据应在正常业务生命周期内完成保存或使用独立的可靠性机制。
Entity 与状态时序
异步系统回调返回协程后,底层调用链不会等待协程完成,也不会使用协程最终的返回值。协程每次 await 之后,以下状态都可能已经变化:
| 状态 | 可能的变化 |
|---|---|
| Entity 生命周期 | Entity 已销毁、迁移、失去 Cell 或客户端断开。 |
| Space 和位置 | Entity 已进入其他 Space,之前的坐标或 Witness 关系失效。 |
| 请求版本 | 新请求已经替代旧请求,旧响应晚到。 |
| 权限和会话 | 登录态、权限或绑定关系已变更。 |
| 业务数据 | 属性已被其他回调修改,恢复后继续写入会覆盖新状态。 |
推荐在发起异步操作前保存 Entity ID、请求版本和必要的不可变数据;恢复后重新取得当前对象并验证版本,只允许仍然有效的请求提交结果。
import asyncio
import KBEngine
async def refresh_profile(entity_id: int, request_version: int) -> None:
profile = await request_profile_from_service(entity_id)
# await 之后重新查找 Entity,不能假设原对象仍然有效。
# Resolve the Entity again after await; the original object may no longer be valid.
entity = KBEngine.entities.get(entity_id)
if entity is None:
return
# 只允许最新请求更新状态,避免迟到响应覆盖新结果。
# Only the newest request may update state, preventing stale responses from overwriting new results.
if entity.profileRequestVersion != request_version:
return
entity.applyProfile(profile)对于 CellApp 的移动、Space 切换、Trap、Witness 和导航回调,要特别注意 await 前后的空间状态不再连续。需要严格顺序的逻辑应保持同步,或者使用明确的请求状态机。
阻塞工作与线程
asyncio.to_thread() 可以把同步阻塞函数移到 Python 线程池,适合少量无法替换的同步 IO,但不能把 Entity 操作放入工作线程。
import asyncio
from pathlib import Path
def read_file(path: Path) -> bytes:
# 此函数只处理普通数据,不访问 KBEngine Entity 或组件对象。
# This function handles plain data only and does not access KBEngine Entity or component objects.
return path.read_bytes()
async def load_file(path: Path) -> bytes:
data = await asyncio.to_thread(read_file, path)
# await 返回后已回到组件主线程,可以在这里应用结果。
# Execution resumes on the component main thread, so results may be applied here.
return data使用线程时需要注意:
| 风险 | 说明 |
|---|---|
| Entity 线程安全 | 工作线程不得读取或修改 Entity、Space、EntityCall、Watcher 等组件状态。 |
| GIL | 纯 Python CPU 密集计算通常不会因 to_thread() 获得理想的并行加速。 |
| 内存 | Python 默认线程池会增加线程栈、任务队列和结果缓存,不等同于引擎 thread_pool 配置。 |
| 关闭 | 进程退出前仍应控制任务数量和超时,不能依赖线程无限等待。 |
| 结果大小 | 大对象在线程和主线程间传递仍会产生内存、GC 和应用结果的 Tick 压力。 |
CPU 密集计算更适合拆批、使用释放 GIL 的成熟原生库,或迁移到职责明确的独立服务。不要为了隐藏慢 Tick 而无限扩大线程池。
aiohttp 服务示例
HTTP 服务更适合部署在 Interfaces 或独立网关中。若确实需要在 BaseApp 中运行,应显式管理 AppRunner 生命周期。
安装依赖
先按 VENV 虚拟环境 配置项目环境,再安装 aiohttp:
pip install aiohttp创建 HTTP 模块
在 server_common 中创建 http_server.py:
from aiohttp import web
import KBEngine
_runner: web.AppRunner | None = None
async def handle_root(request: web.Request) -> web.Response:
return web.Response(text="Hello KBEngine HTTP Server")
async def get_entities(request: web.Request) -> web.Response:
# 生产接口应做认证、分页和返回数量限制,避免一次遍历及序列化过多 Entity。
# Production endpoints need authentication, pagination, and response limits to avoid oversized Entity scans.
entities = [
{"id": entity_id, "class": entity.__class__.__name__}
for entity_id, entity in list(KBEngine.entities.items())[:100]
]
return web.json_response({"count": len(entities), "entities": entities})
async def start_http_server(host: str = "127.0.0.1", port: int = 8001) -> None:
global _runner
if _runner is not None:
return
app = web.Application()
app.router.add_get("/", handle_root)
app.router.add_get("/entities", get_entities)
runner = web.AppRunner(app)
await runner.setup()
try:
site = web.TCPSite(runner, host, port)
await site.start()
except Exception:
await runner.cleanup()
raise
# 保留 runner,以便关服时关闭监听 Socket 和活动连接。
# Retain the runner so listening sockets and active connections can be closed during shutdown.
_runner = runner
async def stop_http_server() -> None:
global _runner
runner = _runner
_runner = None
if runner is not None:
await runner.cleanup()在入口回调中启动
from KBEDebug import INFO_MSG
from http_server import start_http_server
async def onBaseAppReady(is_bootstrap):
# 系统回调直接返回协程时,引擎会创建并推进对应 Task。
# When a system callback returns a coroutine, the engine creates and pumps its Task.
INFO_MSG("onBaseAppReady: isBootstrap=%s" % is_bootstrap)
if is_bootstrap:
await start_http_server()
INFO_MSG("aiohttp HTTP server started at http://127.0.0.1:8001")启动服务后访问 http://127.0.0.1:8001/。正式部署必须增加认证、请求体上限、并发限制、超时和反向代理,不要把 Entity 调试接口直接暴露到公网。
系统回调支持范围
带返回值、返回值必须立即参与底层决策,或者要求同步完成的回调不能改成 async def。支持异步的回调,其 coroutine 结果只表示任务完成,不会重新进入原 C++ 调用链作为业务返回值。
重要限制
不要在不了解回调时序和 Entity 生命周期时把回调批量改成 async def。
带有同步返回值契约的回调一律不能使用 async。CellApp 中与时间、空间、迁移和导航相关的回调尤其要验证 await 前后的状态。
Interfaces
| 对象 | 支持异步的回调 |
|---|---|
KBEngine | onInterfaceAppReady、onRequestCreateAccount、onRequestAccountLogin、onRequestCharge、onInit、onInterfaceAppShutDown |
BaseApp
| 对象 | 支持异步的回调 | 不支持异步的回调 |
|---|---|---|
KBEngine | onGlobalDataDel、onGlobalData、onBaseAppDataDel、onBaseAppData、onInit、onCellAppDeath、onLoseChargeCB、onAutoLoadEntityCreate、onBaseAppReady、onBaseAppShutDown | onReadyForLogin、onReadyForShutDown |
Entity | onDestroy、onCreateCellFailure、onGetCell、onClientDeath、onLoseCell、onRestore、onWriteToDB、onTeleportFailure、onTeleportSuccess、onTimer | onPreArchive |
Proxy | onClientEnabled、onClientDeath、onClientGetCell、onGiveClientToFailure、onStreamComplete | onLogOnAttempt |
EntityComponent | onAttached、onDetached | - |
CellApp
| 对象 | 支持异步的回调 | 不支持异步的回调 |
|---|---|---|
KBEngine | onGlobalDataDel、onGlobalData、onCellAppDataDel、onCellAppData、onInit | - |
Entity | onDestroy、onSpaceGone、onWriteToDB、onWitnessed、onLoseControlledBy、onEnterTrap、onLeaveTrap、onEnteredView、onGetWitness、onLoseWitness、onMove、onMoveOver、onMoveFailure、onTurn、onTeleport、onTeleportFailure、onTeleportSuccess、onEnterSpace、onLeaveSpace、onEnteredCell、onEnteringCell、onLeavingCell、onLeftCell、onRestore、onTimer、onSpaceGeometryLoaded、onAllSpaceGeometryLoaded、onSpaceData、onUpdateBegin、onUpdateEnd | onReadyForLogin |
LoginApp
| 支持异步的回调 | 不支持异步的回调 |
|---|---|
onLoginAppShutDown、onLoseLogin、onLoginAppReady、onCreateAccountCallbackFromDB、onLoginCallbackFromDB | onRequestCreateAccount、onRequestLogin |
Bots
| 支持异步的回调 |
|---|
onInit、onFinish |
生产检查表
| 检查项 | 通过标准 |
|---|---|
| 配置 | 使用 asyncio 的组件已设置正数 asyncioRepeatOffset,并通过压测确定间隔。 |
| 非阻塞性 | 协程中没有 time.sleep()、同步 HTTP、同步文件或同步数据库调用。 |
| 单次执行 | 两次 await 之间没有大循环、大序列化或无限制 Entity 遍历。 |
| 并发 | 外部请求、后台任务和返回结果处理都有上限。 |
| 超时 | 每个外部调用都有明确超时和错误处理。 |
| 生命周期 | 后台 Task 有强引用、异常回收、取消和清理路径。 |
| Entity 状态 | await 后重新验证 Entity、请求版本、Space 和会话状态。 |
| 关服 | 不依赖关服回调完成长时间外部 IO;重要状态有独立可靠性保证。 |
| 性能 | 观察 Tick p95/p99、Task 数量、RSS、GC、网络连接数和外部服务延迟。 |
