From 710dd17e2e8c817acd320f962ceea58189e74ee0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=A4=8F=E6=A5=9A?= <771730766@qq.com> Date: Fri, 10 Apr 2026 17:00:44 +0800 Subject: [PATCH] =?UTF-8?q?Created=20ZLMediaKit=20Python=E6=B7=B7=E5=90=88?= =?UTF-8?q?=E7=BC=96=E7=A8=8B=E6=A8=A1=E5=BC=8F=20(markdown)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ZLMediaKit-Python混合编程模式.md | 579 +++++++++++++++++++++++++++++++ 1 file changed, 579 insertions(+) create mode 100644 ZLMediaKit-Python混合编程模式.md diff --git a/ZLMediaKit-Python混合编程模式.md b/ZLMediaKit-Python混合编程模式.md new file mode 100644 index 0000000..317cf6a --- /dev/null +++ b/ZLMediaKit-Python混合编程模式.md @@ -0,0 +1,579 @@ +> 本文面向对流媒体开发感兴趣的工程师，介绍 ZLMediaKit 的 **Python 混合编程模式**——用Python监听ZLM内核事件、用 Python 编写HTTP API，两件事只需一个 `.py` 文件，开箱即用。 + +--- + +## 一、环境准备：开启 Python 插件功能 + +在写代码之前，有三件事需要先确认。 + +### 1.1 使用支持 Python 的 ZLM 二进制包 + +ZLM 的 Python 插件功能需要在**编译时**开启 `-DENABLE_PYTHON=ON`，普通官方发布包默认不含此功能。 + +有两种方式获取支持 Python 的二进制包： + +**方式一（推荐）：下载预编译包** + +前往 ZLMediaKit 的 [Issue #483](https://github.com/ZLMediaKit/ZLMediaKit/issues/483) 页面，下载文件名中带 `Python` 后缀的二进制包，例如： + +``` +Linux_Python_master_2026-04-01.zip +Windows_Python_master_2026-04-01.zip +``` + +**方式二：自行编译** + +```bash +git clone https://github.com/ZLMediaKit/ZLMediaKit.git +git submodule update --init --recursive +cd ZLMediaKit +mkdir build && cd build +cmake .. -DENABLE_PYTHON=ON +make -j$(nproc) +``` + +--- + +### 1.2 配置 `config.ini` + +在 ZLM 的配置文件 `config.ini` 中添加 `[python]` 节，指定插件入口模块名（不含 `.py` 后缀）： + +```ini +[python] +plugin=mk_plugin +``` + +> `plugin` 的值是 Python 模块名，ZLM 启动时会执行 `import mk_plugin`，因此该模块必须在 Python 的导入路径中可见。 + +--- + +### 1.3 设置 `PYTHONPATH` + +ZLM 需要知道去哪里找 `mk_plugin.py`，启动前设置环境变量 `PYTHONPATH` 指向插件文件所在目录： + +```bash +# 假设 mk_plugin.py 位于 /opt/pymkui/backend/ +export PYTHONPATH=/opt/pymkui/backend + +# 然后启动 ZLM +./MediaServer -c config.ini +``` + +也可以写成一行： + +```bash +PYTHONPATH=/opt/pymkui/backend ./MediaServer -c config.ini +``` + +如果使用 `systemd` 管理服务，在 `[Service]` 段加入： + +```ini +[Service] +Environment="PYTHONPATH=/opt/pymkui/backend" +ExecStart=/opt/zlm/MediaServer -c /opt/zlm/config.ini +``` + +--- + +### 1.4 验证是否生效 + +ZLM 启动日志中出现以下内容，说明 Python 插件加载成功： + +```bash +2026-04-10 16:45:38.811 I [MediaServer] [39041-8660253376] pyinvoker.cpp:560 set_python_path | PYTHONPATH is already set to: /opt/pymkui/backend +2026-04-10 16:45:39.161 I [MediaServer] [39041-8660253376] mk_plugin.py:100 on_start | on_start, secret: xxxxxxxxxxxxxxx +``` + +若出现 `ModuleNotFoundError: No module named 'mk_plugin'`，请检查 `PYTHONPATH` 是否设置正确。 + +--- + +## 二、什么是"混合编程"？ + +ZLMediaKit（简称 ZLM）的核心是高性能 C++ 引擎，但它内置了一套 Python 插件机制（`mk_loader`）： + +``` +┌───────────────────────────────────┐ +│ ZLMediaKit C++ 内核 │ +│ │ +│ 推流 / 拉流 / 转码 / RTP / ... │ +│ │ +│ ┌──────────────────────────────┐ │ +│ │ Python 插件层（mk_loader） │ │ +│ │ │ │ +│ │ 事件回调 ←→ mk_plugin.py │ │ +│ │ HTTP 接口 ←→ FastAPI app │ │ +│ └──────────────────────────────┘ │ +└───────────────────────────────────┘ +``` + +- **事件回调**：推流、播放、录制、流无人观看……C++ 内核触发事件后自动调用 Python 函数，你可以在这里写鉴权、写数据库、发通知。 +- **HTTP 接口扩展**：把一个标准的 `FastAPI` 应用传给 ZLM，ZLM的HTTP服务器会把请求"转发"给你的Python路由，完全复用ZLM的HTTP端口，无需另起进程。 + +--- + +## 三、插件入口文件 `mk_plugin.py` + +ZLM 启动时会加载 `mk_plugin.py`，约定以下几个生命周期函数： + +| 函数名 | 触发时机 | +| ---------------------- | ---------------------- | +| `on_start()` | ZLM 启动完成后调用一次 | +| `on_exit()` | ZLM 退出前调用一次 | +| `on_reload_config()` | 配置热重载时触发 | + +所有**事件回调**也定义在同一个文件里，返回 `True` 代表 Python 已接管该事件，ZLM 不再做默认处理；返回 `False` 则继续走 ZLM默认WebHook逻辑。 + +--- + +## 四、常用事件回调详解 + +### 4.1 `on_start` — 启动初始化 + +```python +import mk_loader +import mk_logger + +def on_start(): + # 读取/修改 ZLM 配置项 + mk_loader.set_config('http.rootPath', '/var/www/frontend') + mk_loader.update_config() # 使配置生效 + + # 把 FastAPI 路由注册到 ZLM HTTP 服务器 + mk_loader.set_fastapi(check_route, submit_coro) + + mk_logger.log_info("插件启动完成") +``` + +> `set_fastapi` 传入两个函数：`check_route`（判断路径是否由 Python 处理）和 `submit_coro`（把 ASGI 请求投递到异步事件循环）。这两个函数是固定写法，下文会给出完整示例。 + +--- + +### 4.2 `on_publish` — 推流鉴权 + +每当有客户端推流时触发，可在此做 Token 校验、写入流记录等。 + +```python +def on_publish(type: str, args: dict, invoker, sender: dict) -> bool: + stream = args.get("stream", "") + app = args.get("app", "") + token = args.get("params", "") # URL 参数，如 ?token=xxx + + # 简单示例：校验 token + if not is_valid_token(token, stream): + # 第二个参数非空字符串 = 拒绝推流，内容作为错误信息返回给推流端 + mk_loader.publish_auth_invoker_do(invoker, "Token 无效，推流拒绝") + return True + + # 允许推流，同时指定转协议选项 + opt = { + "enable_hls": "1", # 开启 HLS + "enable_mp4": "0", # 不自动录制 + } + mk_loader.publish_auth_invoker_do(invoker, "", opt) + mk_logger.log_info(f"推流已允许: {app}/{stream}") + return True +``` + +--- + +### 4.3 `on_play` — 播放鉴权 + +```python +def on_play(args: dict, invoker, sender: dict) -> bool: + stream = args.get("stream", "") + ip = sender.get("peer_ip", "") + + # 白名单校验示例 + if ip not in ALLOWED_IPS: + mk_loader.play_auth_invoker_do(invoker, "IP 不在白名单") + return True + + mk_loader.play_auth_invoker_do(invoker, "") # 空字符串 = 允许 + return True +``` + +--- + +### 4.4 `on_stream_not_found` — 按需拉流 + +这是混合编程最典型的使用场景：播放器请求一条不存在的流时，ZLM 触发此事件，Python 可以在这里动态启动拉流代理。 + +```python +def on_stream_not_found(args: dict, sender: dict, invoker) -> bool: + vhost = args.get("vhost", "__defaultVhost__") + app = args.get("app", "") + stream = args.get("stream", "") + + # 查数据库：是否存在该流的"按需拉流"配置 + proxy = db.get_proxy(vhost, app, stream, on_demand=True) + if not proxy: + return False # 不处理，ZLM 返回 404 + + url = proxy["url"] + + def cb(err, key): + if err: + mk_logger.log_warn(f"按需拉流失败: {err}") + else: + mk_logger.log_info(f"按需拉流成功: {key}") + + mk_loader.add_stream_proxy( + vhost, app, stream, url, + cb, + retry_count=3, + force=True, + timeout_sec=10.0, + opt={"auto_close": True}, # 无人观看后自动停止 + ) + return True # 告诉 ZLM：Python 已接管，等待流上线再分发给播放器 +``` + +--- + +### 4.5 `on_stream_none_reader` — 无人观看 + +```python +def on_stream_none_reader(sender: mk_loader.MediaSource) -> bool: + url = sender.getUrl() + mk_logger.log_info(f"无人观看: {url}") + + # 可以在这里选择关闭流，释放带宽 + # sender.close(False) + + return True # 返回 True 阻止 ZLM 默认行为（默认也是关流） +``` + +--- + +### 4.6 `on_record_mp4` / `on_record_ts` — 录制完成 + +```python +def on_record_mp4(info: dict) -> bool: + """ + info 包含： + file_path — 录制文件本地路径 + start_time — 录制开始时间戳 + time_len — 时长（秒） + file_size — 文件大小（字节） + vhost / app / stream + """ + mk_logger.log_info(f"录制完成: {info['file_path']} ({info['time_len']}s)") + + # 异步上传到 OSS / 发送 Webhook 通知等 + asyncio.run_coroutine_threadsafe( + upload_to_oss(info["file_path"]), + SharedLoop.get_loop() + ) + return True +``` + +--- + +### 4.7 `on_player_proxy_failed` — 拉流失败自动切换备用地址 + +```python +def on_player_proxy_failed(url, media_tuple, ex) -> bool: + vhost = media_tuple.vhost + app = media_tuple.app + stream = media_tuple.stream + + # 从数据库取多条备用地址，切换到下一个 + next_url = db.get_next_proxy_url(vhost, app, stream, failed_url=url) + if not next_url: + return False + + mk_logger.log_info(f"切换备用地址: {url} → {next_url}") + mk_loader.update_stream_proxy(vhost, app, stream, next_url, {}) + return True +``` + +--- + +### 4.8 `on_http_access` — HTTP 文件访问鉴权 + +```python +def on_http_access(parser, path, file_path, is_dir, invoker, sender) -> bool: + frontend_root = "/var/www/frontend" + + if not file_path.startswith(frontend_root): + # 禁止访问前端目录之外的文件，防止路径穿越 + mk_loader.http_access_invoker_do(invoker, "Access denied", path, 3600) + return True + + mk_loader.http_access_invoker_do(invoker, "", path, 3600) + return True +``` + +--- + +## 五、用 Python 扩展 ZLM 的 HTTP 接口 + +### 5.1 核心原理 + +ZLM的HTTP服务器收到请求后，先查询Python是否注册了对应路由；如果有，则把请求以 **ASGI** 格式投递到Python的 `FastAPI` 应用。整个流程在同一进程内完成，没有网络开销。 + +### 5.2 固定胶水代码（`mk_plugin.py`） + +```python +import asyncio +from starlette.routing import Match +from shared_loop import SharedLoop +from py_http_api import app # 你的 FastAPI 实例 + +def submit_coro(scope, body, send): + """把 ZLM 传来的 ASGI 请求投递到 Python 异步事件循环""" + async def run(): + async def async_send(message): + result = send(message) + if result is not None: + await result + + async def receive(): + return {"type": "http.request", "body": body, "more_body": False} + + try: + await app(scope, receive, async_send) + except Exception as e: + mk_logger.log_warn(f"FastAPI 处理失败: {e}") + await async_send({"type": "http.response.start", "status": 500, + "headers": [(b"content-type", b"text/plain")]}) + await async_send({"type": "http.response.body", "body": b"Internal Server Error"}) + + return asyncio.run_coroutine_threadsafe(run(), SharedLoop.get_loop()) + + +def check_route(scope) -> bool: + """告诉 ZLM 这个路径是否由 Python 处理""" + for route in app.routes: + if hasattr(route, "matches"): + match, _ = route.matches(scope) + if match == Match.FULL: + return True + return False + + +def on_start(): + mk_loader.set_fastapi(check_route, submit_coro) +``` + +### 5.3 编写 FastAPI 接口（`py_http_api.py`） + +接口与普通 FastAPI 项目完全一致，支持 `GET`/`POST`、路径参数、Body 解析等： + +```python +import json +import mk_logger +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +app = FastAPI(title="我的流媒体接口") + +# ── 示例 1：查询在线流列表 ────────────────────────────────────────── +@app.get("/index/pyapi/streams") +async def list_streams(): + streams = db.get_all_streams() + return {"code": 0, "data": streams} + + +# ── 示例 2：添加拉流代理（POST，支持 JSON / Form 两种请求体） ──────── +@app.post("/index/pyapi/addPullProxy") +async def add_pull_proxy(request: Request): + # 解析请求体 + body = await request.body() + content_type = request.headers.get("content-type", "") + if "application/json" in content_type: + data = json.loads(body) + else: + import urllib.parse + parsed = urllib.parse.parse_qs(body.decode(), keep_blank_values=True) + data = {k: v[0] for k, v in parsed.items()} + + app_name = data.get("app", "") + stream = data.get("stream", "") + url = data.get("url", "") + + if not all([app_name, stream, url]): + return {"code": -1, "msg": "app/stream/url 不能为空"} + + proxy_id = db.add_pull_proxy({ + "vhost": "__defaultVhost__", + "app": app_name, + "stream": stream, + }) + db.set_proxy_urls(proxy_id, [{"url": url, "params": {}}]) + + mk_logger.log_info(f"添加拉流代理: {app_name}/{stream} → {url}") + return {"code": 0, "msg": "添加成功", "data": {"id": proxy_id}} + + +# ── 示例 3：调用 ZLM 原生 API（反向代理模式） ──────────────────────── +import httpx + +_client = httpx.AsyncClient(timeout=10.0) + +@app.get("/index/pyapi/zlm/mediaList") +async def get_media_list(request: Request): + """把请求透传给 ZLM 原生接口，做一层 Python 加工后返回""" + secret = mk_loader.get_config("api.secret") + zlm_url = f"http://127.0.0.1:{mk_loader.get_config('http.port')}/index/api/getMediaList" + resp = await _client.get(zlm_url, params={"secret": secret}) + data = resp.json() + + # 在这里可以对数据做过滤、脱敏等处理 + return {"code": 0, "total": len(data.get("data", [])), "data": data.get("data", [])} +``` + +### 5.4 接口路径约定 + +| 前缀 | 说明 | +| -------------------- | ------------------------------- | +| `/index/api/...` | ZLM 原生接口，直接调用 | +| `/index/pyapi/...` | Python 扩展接口（推荐命名规范） | + +--- + +## 六、异步与线程安全 + +ZLM 的 C++ 回调运行在 **C++ 线程**里，而 FastAPI 运行在 **Python asyncio 事件循环**里。两者之间的桥梁是 `SharedLoop`： + +```python +# shared_loop.py — 全局共享的 asyncio 事件循环 +import asyncio +import threading + +class SharedLoop: + _loop: asyncio.AbstractEventLoop = None + _thread: threading.Thread = None + + @classmethod + def get_loop(cls) -> asyncio.AbstractEventLoop: + if cls._loop is None: + cls._loop = asyncio.new_event_loop() + cls._thread = threading.Thread( + target=cls._loop.run_forever, daemon=True + ) + cls._thread.start() + return cls._loop +``` + +在 C++ 回调函数里投递异步任务： + +```python +import asyncio +from shared_loop import SharedLoop + +def on_record_mp4(info: dict) -> bool: + # ✅ 正确：通过 run_coroutine_threadsafe 跨线程投递协程 + asyncio.run_coroutine_threadsafe( + notify_webhook(info), + SharedLoop.get_loop() + ) + return True + +async def notify_webhook(info: dict): + async with httpx.AsyncClient() as client: + await client.post("https://your-server.com/hook", json=info) +``` + +> ⚠️ 不要在 C++ 回调里直接 `await`，也不要直接调用 `asyncio.run()`，否则会发生死锁或运行时错误。 + +--- + +## 七、完整插件文件骨架 + +```python +# mk_plugin.py — ZLMediaKit Python 插件完整骨架 + +import asyncio +import mk_loader +import mk_logger +from starlette.routing import Match +from shared_loop import SharedLoop +from py_http_api import app # FastAPI 实例 + +# ── ASGI 桥接（固定写法） ───────────────────────────────────────── +def submit_coro(scope, body, send): + async def run(): + async def async_send(msg): + r = send(msg) + if r is not None: + await r + async def receive(): + return {"type": "http.request", "body": body, "more_body": False} + try: + await app(scope, receive, async_send) + except Exception as e: + mk_logger.log_warn(f"FastAPI error: {e}") + return asyncio.run_coroutine_threadsafe(run(), SharedLoop.get_loop()) + +def check_route(scope) -> bool: + for route in app.routes: + if hasattr(route, "matches"): + match, _ = route.matches(scope) + if match == Match.FULL: + return True + return False + +# ── 生命周期 ────────────────────────────────────────────────────── +def on_start(): + mk_loader.set_config("http.rootPath", "/var/www/frontend") + mk_loader.update_config() + mk_loader.set_fastapi(check_route, submit_coro) + mk_logger.log_info("Python 插件已启动") + +def on_exit(): + mk_logger.log_info("Python 插件退出") + +def on_reload_config(): + mk_logger.log_info("配置已热重载") + +# ── 推流鉴权 ────────────────────────────────────────────────────── +def on_publish(type: str, args: dict, invoker, sender: dict) -> bool: + mk_loader.publish_auth_invoker_do(invoker, "", {}) + return True + +# ── 播放鉴权 ────────────────────────────────────────────────────── +def on_play(args: dict, invoker, sender: dict) -> bool: + mk_loader.play_auth_invoker_do(invoker, "") + return True + +# ── 按需拉流 ────────────────────────────────────────────────────── +def on_stream_not_found(args: dict, sender: dict, invoker) -> bool: + # 查询数据库 → 动态启动拉流代理 + return False # 没有匹配时返回 False + +# ── 录制完成 ────────────────────────────────────────────────────── +def on_record_mp4(info: dict) -> bool: + mk_logger.log_info(f"录制: {info.get('file_path')} {info.get('time_len')}s") + return True + +# ── 无人观看 ────────────────────────────────────────────────────── +def on_stream_none_reader(sender: mk_loader.MediaSource) -> bool: + return True + +# ── HTTP 文件访问鉴权 ───────────────────────────────────────────── +def on_http_access(parser, path, file_path, is_dir, invoker, sender) -> bool: + mk_loader.http_access_invoker_do(invoker, "", path, 3600) + return True +``` + +--- + +## 八、总结 + +| 能力 | 实现方式 | +| ----------------- | ------------------------------------------------------------------ | +| 推流 / 播放鉴权 | `on_publish` / `on_play` + `invoker_do` | +| 按需拉流 | `on_stream_not_found` + `mk_loader.add_stream_proxy` | +| 拉流多地址切换 | `on_player_proxy_failed` + `mk_loader.update_stream_proxy` | +| 录制后处理 | `on_record_mp4` + 异步投递到 SharedLoop | +| 扩展 HTTP API | FastAPI 路由 +`set_fastapi` 注册 | +| 调用 ZLM 原生 API | `httpx.AsyncClient` → `http://127.0.0.1:{port}/index/api/...` | +| 跨线程异步 | `asyncio.run_coroutine_threadsafe(..., SharedLoop.get_loop())` | + +ZLMediaKit的Python插件机制把 **C++ 的高性能**和 **Python 的开发效率**结合在一起：核心推拉流由 C++ 保证吞吐量，业务逻辑、数据库操作、HTTP 接口全用 Python 快速迭代，是流媒体服务开发的一种高效范式。 + +--- + +*本文代码均来自开源项目 [PyMKUI](https://github.com/ZLMediaKit/pymkui)，欢迎 Star ⭐* \ No newline at end of file