NcatBot 文档

外观

加载与卸载流程

约 1904 字大约 6 分钟

2026-03-19

插件从发现到就绪的完整生命周期——扫描、依赖排序、导入、Mixin 钩子链、卸载与清理。

目录

全流程概览

加载阶段

1. 扫描

PluginIndexer 递归扫描 plugins/ 目录，查找所有包含 manifest.toml 的子目录：

plugins/
├── hello_world/manifest.toml  ✅ 发现
├── my_plugin/manifest.toml    ✅ 发现
└── no_manifest/               ❌ 跳过

解析 manifest.toml 为 PluginManifest 对象，验证必填字段和入口文件是否存在。

2. 依赖拓扑排序

DependencyResolver 使用 Kahn 算法（拓扑排序）确定加载顺序：

  • 根据 manifest.toml 中的 [dependencies] 构建有向依赖图
  • 无依赖的插件先加载，被依赖的插件保证在依赖方之前加载
  • 检测循环依赖：如果存在 A → B → A 的环，会抛出 PluginCircularDependencyError
  • 检测缺失依赖：如果依赖的插件不存在，会抛出 PluginMissingDependencyError
  • 使用 packaging.specifiers 验证版本约束

3. 导入模块

ModuleImporter 使用 importlib 动态导入插件入口模块：

  • 插件根目录被添加到 sys.path（低优先级，不影响标准库和第三方包）
  • 导入前自动清理 __pycache__，确保代码更新生效
  • 使用 ContextVar 隔离当前加载插件的名称（用于装饰器注册 Handler 时标记归属）
  • 如果 __init__.py 中含 from .main import ...，Python import system 会先于 load_module() 导入入口模块。框架会检测到入口模块已在 sys.modules 中，直接复用而不重新执行，避免装饰器注册出重复的 Handler

4. 实例化 + 属性注入

PluginLoader._instantiate() 创建插件实例并注入运行时属性：

plugin.workspace = plugin_workspace_path
plugin.services = service_manager
plugin.api = bot_api_client
plugin._dispatcher = event_dispatcher
plugin._plugin_loader = self
plugin._manifest = manifest
plugin._debug = debug_flag

5-8. __onload__() 编排

框架调用 plugin.__onload__()，该方法按顺序执行：

async def __onload__(self) -> None:
    self.workspace.mkdir(exist_ok=True, parents=True)  # 5. 创建工作目录
    await self._run_mixin_hooks("_mixin_load")          # 6. Mixin 加载钩子
    self._init_()                                        # 7. 同步预初始化
    await self.on_load()                                 # 8. 异步主初始化

9. 刷新 Handler

通过 @registrar.on_*() 装饰器声明的 Handler 会被暂存，__onload__() 完成后一次性刷新到 HandlerDispatcher。因此在 on_load() 中，本插件的 Handler 尚未全部就绪。

10. 执行就绪回调

通过 @registrar.on_startup() 装饰的插件实例异步方法会在加载完成后执行。首次启动时，框架先加载所有插件并刷新 Handler，再运行这些回调；运行中加载或热重载插件时，在该插件完成加载后运行它的回调。回调抛出异常时框架记录日志，不阻断插件加载；热重载会重新触发回调。

Mixin 钩子链

NcatBotPlugin 继承链中的每个 Mixin 都可以定义 _mixin_load() 和 _mixin_unload() 钩子。框架按 MRO（方法解析顺序） 自动发现并依次执行。

执行顺序

NcatBotPlugin(BasePlugin, EventMixin, TimeTaskMixin, RBACMixin, ConfigMixin, DataMixin)
顺序Mixin_mixin_load() 做什么_mixin_unload() 做什么
1EventMixin初始化事件流列表关闭所有活跃的 EventStream
2TimeTaskMixin初始化任务名列表清理所有定时任务
3RBACMixin（无特殊操作）（无特殊操作）
4ConfigMixin从 config.yaml 加载配置保存配置到 config.yaml
5DataMixin从 data.json 加载数据保存数据到 data.json

独立容错

每个 Mixin 钩子在独立的 try/except 中执行——单个 Mixin 失败不会阻止其他 Mixin 初始化：

async def _run_mixin_hooks(self, hook_name: str):
    for cls in type(self).__mro__:
        hook = cls.__dict__.get(hook_name)
        if hook is not None:
            try:
                result = hook(self)
                if asyncio.iscoroutine(result):
                    await result
            except Exception:
                LOG.exception("Mixin hook %s.%s 执行失败", cls.__name__, hook_name)

卸载阶段

__unload__() 编排

async def __unload__(self) -> None:
    self._close_()                                       # 11. 同步后清理
    await self.on_close()                                # 12. 异步清理
    await self._run_mixin_hooks("_mixin_unload")          # 13. Mixin 卸载钩子

Handler 撤销

卸载后，HandlerDispatcher.revoke_plugin(name) 移除该插件注册的所有 Handler，确保不会再响应事件。

模块清理

ModuleImporter.unload_module() 从 sys.modules 中移除插件相关的模块条目，释放旧代码引用。

开发者钩子 API

作为插件开发者，你可以重写以下 4 个生命周期钩子，并使用 @registrar.on_startup() 声明就绪回调：

钩子类型调用时机典型用途
_init_(self)同步Mixin 加载后、on_load() 之前同步初始化（较少使用）
on_load(self)异步单个插件加载期间初始化数据、调用 Bot API、启动后台任务
on_close(self)异步卸载时清理资源、保存状态
_close_(self)同步on_close() 之前同步清理（较少使用）
装饰器回调签名调用时机典型用途
@registrar.on_startup()async def callback(self)首次启动时所有插件及 Handler 就绪后；运行中加载或热重载时该插件加载完成后依赖已注册 Handler 或其他已加载插件的启动操作

on_load() 在单个插件的加载过程中执行：self.api 和 self.services 已可用，适合初始化配置、注册权限或调用 Bot API；但本插件的 Handler 尚未刷新，后续插件也可能尚未加载。@registrar.on_startup() 在加载完成后执行，适合需要等待 Handler 就绪的操作。首次启动时可访问全部已加载插件；运行中加载或重载时，仅保证当前插件已完成加载。两者都会在热重载时再次执行。

插件的 @registrar.on_startup() 与非插件模式的 @bot.on_startup() 是不同入口。前者装饰插件类的实例异步方法，按插件加载周期运行；后者装饰 BotClient 回调，随 Bot 启动运行。

最常用的是 on_load() 和 on_close()：

class MyPlugin(NcatBotPlugin):
    name = "my_plugin"
    version = "1.0.0"

    async def on_load(self):
        if not self.get_config("prefix"):
            self.set_config("prefix", "/")
        self.data.setdefault("counter", 0)
        self.add_scheduled_task("heartbeat", "60s")
        LOG.info("MyPlugin 已加载")

    async def on_close(self):
        LOG.info("MyPlugin 已卸载，累计计数: %d", self.data.get("counter", 0))

完整示例：examples/common/02_config_and_data/ 展示了在 on_load() 中初始化配置和数据。

下面的插件在 on_load() 中读取 API，并在插件就绪后记录登录信息：

from ncatbot.core import registrar
from ncatbot.plugin import NcatBotPlugin


class MyPlugin(NcatBotPlugin):
    name = "my_plugin"
    version = "1.0.0"

    async def on_load(self):
        self.login_info = await self.api.qq.query.get_login_info()

    @registrar.on_startup()
    async def ready(self):
        self.logger.info("插件已就绪，登录信息：%s", self.login_info)

常见模式

在 on_load() 中注册事件处理器

事件处理装饰器声明的方法会在类定义时收集，在 on_load() 完成后自动刷新到分发器。通常不需要在 on_load() 中手动操作 Handler。@registrar.on_startup() 用于就绪回调，不会注册事件 Handler。

在 on_load() 中启动后台任务

如果需要持续运行的后台任务，在 on_load() 中使用 asyncio.create_task()，并在 on_close() 中取消：

class MyPlugin(NcatBotPlugin):
    name = "my_plugin"
    version = "1.0.0"

    async def on_load(self):
        self._task = asyncio.create_task(self._background_worker())

    async def on_close(self):
        if hasattr(self, "_task"):
            self._task.cancel()

    async def _background_worker(self):
        try:
            async with self.events("message") as stream:
                async for event in stream:
                    LOG.info("收到消息: %s", event.data.raw_message)
        except asyncio.CancelledError:
            pass

完整示例：examples/qq/01_event_registration/ 的 _stream_listener() 方法。

下一步

版权归属：GEYUANwuqi

许可证：署名 4.0 国际 (CC-BY-4.0)

© 2025-2026 NcatBot