chongming/chongming-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

modify: ruff 清理 + 单例锁持有者释放 + 移除死代码

Browse Source
This commit is contained in:
chakcy
2026-08-14 16:31:37 +08:00
parent 2eb5cc23cb
commit d51633db4d
5 changed files with 67 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/chongming/app/core/lifespan_setup.py
+15 -5
View File
@@ -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)
src/chongming/app/core/mqtt_bridge.py
+47 -22
View File
@@ -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
src/chongming/app/core/worker_manager.py
+1 -2
View File
@@ -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
test_auth_race.py
+1 -1
View File
@@ -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}")
verify_example_plugin.py
+3 -3
View File
@@ -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 = []