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：

xml

<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 调度线程。

text

组件 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` 什么时候会让出执行权

真正的非阻塞等待

python

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` 仍可能阻塞

python

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 连在一起，仍然可能形成一段长时间的连续执行。

以下写法也不适合作为永久循环：

python

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、闭包和结果对象内存，并形成唤醒风暴。

超时与并发限制

外部服务可能永久不返回，生产代码必须设置超时并限制并发量。

python

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() 创建的后台任务，应由项目保留强引用、读取异常并在关闭时取消。

python

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、请求版本和必要的不可变数据；恢复后重新取得当前对象并验证版本，只允许仍然有效的请求提交结果。

python

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 操作放入工作线程。

python

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：

bash

pip install aiohttp

创建 HTTP 模块

在 server_common 中创建 http_server.py：

python

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()

在入口回调中启动

python

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、网络连接数和外部服务延迟。

Space空间-HOWTOs

分布式部署

asyncio 协程

结论

启用 asyncio

引擎执行模型

`await` 什么时候会让出执行权

真正的非阻塞等待

写了 `await` 仍可能阻塞

已完成的 awaitable 可能立即继续

适合与不适合的工作

超时与并发限制

后台任务生命周期

Entity 与状态时序

阻塞工作与线程

aiohttp 服务示例

安装依赖

创建 HTTP 模块

在入口回调中启动

系统回调支持范围

Interfaces

BaseApp

CellApp

LoginApp

Bots

生产检查表

相关文档

asyncio 协程 ​

结论 ​

启用 asyncio ​

引擎执行模型 ​

await 什么时候会让出执行权 ​

真正的非阻塞等待 ​

写了 await 仍可能阻塞 ​

已完成的 awaitable 可能立即继续 ​

适合与不适合的工作 ​

超时与并发限制 ​

后台任务生命周期 ​

Entity 与状态时序 ​

阻塞工作与线程 ​

aiohttp 服务示例 ​

安装依赖 ​

创建 HTTP 模块 ​

在入口回调中启动 ​

系统回调支持范围 ​

Interfaces ​

BaseApp ​

CellApp ​

LoginApp ​

Bots ​

生产检查表 ​

相关文档 ​

asyncio 协程

结论

启用 asyncio

引擎执行模型

`await` 什么时候会让出执行权

真正的非阻塞等待

写了 `await` 仍可能阻塞

已完成的 awaitable 可能立即继续

适合与不适合的工作

超时与并发限制

后台任务生命周期

Entity 与状态时序

阻塞工作与线程

aiohttp 服务示例

安装依赖

创建 HTTP 模块

在入口回调中启动

系统回调支持范围

Interfaces

BaseApp

CellApp

LoginApp

Bots

生产检查表

相关文档