From d51633db4d793c82147d98a36ad98ff001a34a65 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 14 Aug 2026 16:31:37 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20ruff=20=E6=B8=85=E7=90=86=20+=20?= =?UTF-8?q?=E5=8D=95=E4=BE=8B=E9=94=81=E6=8C=81=E6=9C=89=E8=80=85=E9=87=8A?= =?UTF-8?q?=E6=94=BE=20+=20=E7=A7=BB=E9=99=A4=E6=AD=BB=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/chongming/app/core/lifespan_setup.py | 20 +++++-- src/chongming/app/core/mqtt_bridge.py | 69 ++++++++++++++++-------- src/chongming/app/core/worker_manager.py | 3 +- test_auth_race.py | 2 +- verify_example_plugin.py | 6 +-- 5 files changed, 67 insertions(+), 33 deletions(-) diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index 3806314..60f6730 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/src/chongming/app/core/lifespan_setup.py @@ -413,8 +413,13 @@ def setup_mqtt_broker(app: FastAPI): time.sleep(0.1) raise RuntimeError("MQTT Broker 端口未就绪（已被其他 worker 锁定）") + # 记录本进程持有锁，关闭时由持有者释放（避免误删其他 worker 的锁） + app.state.mqtt_broker_lock_acquired = True + if is_port_in_use(port, host): logger.info(f"MQTT Broker 已在 {host}:{port} 运行，跳过") + _release_singleton_lock("mqtt_broker") + app.state.mqtt_broker_lock_acquired = False return proc = multiprocessing.Process( target=_run_mqtt_broker, args=(host, port), daemon=False @@ -433,7 +438,7 @@ def setup_mqtt_broker(app: FastAPI): async def shutdown_mqtt_broker(app: FastAPI): - """终止 MQTT Broker 子进程""" + """终止 MQTT Broker 子进程（仅锁持有者释放锁）""" proc = getattr(app.state, "mqtt_broker_process", None) if proc and proc.is_alive(): proc.terminate() @@ -441,6 +446,8 @@ async def shutdown_mqtt_broker(app: FastAPI): if proc.is_alive(): proc.kill() logger.info("MQTT Broker 子进程已终止") + if getattr(app.state, "mqtt_broker_lock_acquired", False): + _release_singleton_lock("mqtt_broker") async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge | None: @@ -463,13 +470,14 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge custom_topics = list(mqtt_config.get("custom_topics", []) or []) custom_topics += list(getattr(app.state, "plugin_mqtt_topics", []) or []) - # 1. 启动 Broker 子进程（文件锁已确保只有一个） - # setup_mqtt_broker(app, host, port) + # Broker 子进程已在 lifespan 第 3 步启动（setup_mqtt_broker + 文件锁单例） - # 2. 原子锁确保只有一个 worker 启动 MQTT Bridge + # 原子锁确保只有一个 worker 启动 MQTT Bridge if not _acquire_singleton_lock("mqtt_bridge"): logger.info("MQTT Bridge 已由其他 worker 启动，跳过") return None + # 记录本进程持有锁，关闭时由持有者释放（避免误删其他 worker 的锁） + app.state.mqtt_bridge_lock_acquired = True bridge = MQTTBridge( worker_manager=worker_manager, @@ -484,10 +492,12 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge async def shutdown_mqtt(app: FastAPI): - """停止 MQTT Bridge 和 Broker 子进程""" + """停止 MQTT Bridge 和 Broker 子进程（仅锁持有者释放锁）""" bridge: MQTTBridge | None = getattr(app.state, "mqtt_bridge", None) if bridge and bridge.is_running: await bridge.stop() + if getattr(app.state, "mqtt_bridge_lock_acquired", False): + _release_singleton_lock("mqtt_bridge") await shutdown_mqtt_broker(app) diff --git a/src/chongming/app/core/mqtt_bridge.py b/src/chongming/app/core/mqtt_bridge.py index 70b8ddd..c0e2187 100644 --- a/src/chongming/app/core/mqtt_bridge.py +++ b/src/chongming/app/core/mqtt_bridge.py @@ -1,9 +1,13 @@ """MQTT Bridge — 纯客户端模式，Broker 由子进程管理""" -import asyncio, json, re -from typing import Any, Optional +import asyncio +import json +import re +from typing import Optional + from amqtt.client import MQTTClient from amqtt.mqtt.constants import QOS_1 from .logger import get_logger + logger = get_logger("mqtt") MQTT_TOPIC_PREFIX = "chongming" MQTT_SUBSCRIBE_TOPIC = f"{MQTT_TOPIC_PREFIX}/+/+" @@ -34,7 +38,9 @@ def _compile_topic(topic: str) -> re.Pattern: class MQTTBridge: def __init__(self, worker_manager, host="localhost", port=1883, custom_topics=None): - self._wm = worker_manager; self._host = host; self._port = port + self._wm = worker_manager + self._host = host + self._port = port self._client: Optional[MQTTClient] = None self._lt: Optional[asyncio.Task] = None self._running = False @@ -46,15 +52,19 @@ class MQTTBridge: ] async def start(self): - if self._running: return - try: await self._sc() + if self._running: + return + try: + await self._sc() except Exception as e: - logger.warning(f"MQTT 不可用: {e}"); return + logger.warning(f"MQTT 不可用: {e}") + return self._running = True logger.info(f"MQTT Bridge 已连接 {self._host}:{self._port}") async def stop(self): - if not self._running: return + if not self._running: + return self._running = False await self._sc_stop() @@ -73,30 +83,38 @@ class MQTTBridge: if self._client: try: await self._client.disconnect() - except Exception: pass + except Exception: + pass self._client = None # 再取消消息循环 if self._lt and not self._lt.done(): self._lt.cancel() - try: await self._lt - except asyncio.CancelledError: pass + try: + await self._lt + except asyncio.CancelledError: + pass async def _loop(self): while self._running: try: m = await self._client.deliver_message() # type: ignore - if m is None: break + if m is None: + break await self._handle(m) - except asyncio.CancelledError: break + except asyncio.CancelledError: + break except Exception as e: if self._running: logger.error(f"消息异常: {e}") break async def _handle(self, m): - t: str = m.topic; p = m.data # type: ignore - try: pl = json.loads(p.decode("utf-8")) - except json.JSONDecodeError: return + t: str = m.topic + p = m.data # type: ignore + try: + pl = json.loads(p.decode("utf-8")) + except json.JSONDecodeError: + return # 信封协议（严格）: {"meta": {...}, "data": {...}} # 非信封格式载荷直接丢弃并告警，请设备侧按新协议发布 @@ -120,7 +138,8 @@ class MQTTBridge: # 2. 内置接口 chongming/{worker}/{operation} parts = t.split("/") - if len(parts) < 3: return + if len(parts) < 3: + return wt, op = parts[1], parts[2] subj = f"{wt}.{op}" rt = f"{MQTT_TOPIC_PREFIX}/{wt}/{op}/response" @@ -128,7 +147,8 @@ class MQTTBridge: def _fill(self, template: str, matched: re.Match) -> str: """用匹配到的通配符值填充模板中的 {1} {2}... 占位符""" - if "{" not in template: return template + if "{" not in template: + return template for i, g in enumerate(matched.groups(), start=1): template = template.replace("{" + str(i) + "}", g) return template @@ -153,15 +173,20 @@ class MQTTBridge: r = await self._wm.call(worker_type=wt, subject=subj, payload=pl, timeout=float(meta.get("timeout", 30)), cacheable=bool(meta.get("cacheable", False))) - if r is not None: await self._pub(rt, r) - except RuntimeError as e: await self._pub(rt, {"error":str(e)}) - except Exception as e: await self._pub(rt, {"error":str(e)}) + if r is not None: + await self._pub(rt, r) + except RuntimeError as e: + await self._pub(rt, {"error": str(e)}) + except Exception as e: + await self._pub(rt, {"error": str(e)}) async def _pub(self, topic, data): try: pd = json.dumps(data, ensure_ascii=False, default=str) await self._client.publish(topic, pd.encode("utf-8"), QOS_1) # type: ignore - except Exception as e: logger.error(f"发布失败 [{topic}]: {e}") + except Exception as e: + logger.error(f"发布失败 [{topic}]: {e}") @property - def is_running(self): return self._running + def is_running(self): + return self._running diff --git a/src/chongming/app/core/worker_manager.py b/src/chongming/app/core/worker_manager.py index 6a5a6fc..bbfc481 100644 --- a/src/chongming/app/core/worker_manager.py +++ b/src/chongming/app/core/worker_manager.py @@ -2,8 +2,7 @@ import asyncio import random import json import orjson -import time -from typing import Any, Dict, List, Optional, Set +from typing import Any, Dict, List, Optional import zenoh from diskcache import Cache from .logger import get_logger diff --git a/test_auth_race.py b/test_auth_race.py index e7b55f5..e0df485 100644 --- a/test_auth_race.py +++ b/test_auth_race.py @@ -52,7 +52,7 @@ async def main(): res = await sess.execute(select(Users)) rows = res.scalars().all() - print(f"✅ 4 个并发启动全部成功") + print("✅ 4 个并发启动全部成功") print(f"✅ users 表记录数: {len(rows)}（应为 1）") for u in rows: print(f" - {u.username} <{u.email}> superuser={u.is_superuser}") diff --git a/verify_example_plugin.py b/verify_example_plugin.py index 7562f67..f956d6a 100644 --- a/verify_example_plugin.py +++ b/verify_example_plugin.py @@ -16,7 +16,7 @@ def check(name, cond, detail=""): # ========== 1. global_var 白名单 / require / __all__ ========== -from plugins.example import global_var as gv +from plugins.example import global_var as gv # noqa: E402 check("global_var.__all__ 包含 db", "db" in gv.__all__) check("global_var.__all__ 包含 set_global_var", "set_global_var" in gv.__all__) @@ -45,7 +45,7 @@ check("模块属性 db 已被更新", gv.db is marker) # ========== 2. __init__.py 导入链（router + Plugin 基类）========== -import plugins.example as plug +import plugins.example as plug # noqa: E402 check("plugins.example 导入成功", True) check("create_workers 存在", callable(getattr(plug, "create_workers", None))) @@ -54,7 +54,7 @@ check("plugin 实例存在", getattr(plug, "plugin", None) is not None) check("plugin._initialized 默认 False", plug.plugin._initialized is False) # ========== 3. create_workers 幂等性（mock session/config）========== -import asyncio +import asyncio # noqa: E402 # 用桩替代 Plugin 基类的 worker 创建实现，避免真实 zenoh 会话 calls = []