gis/xian_algorithm_new
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

QGIS设置允许并发

Browse Source
This commit is contained in:
wzy-warehouse
2026-06-24 11:22:20 +08:00
parent d402668a5c
commit cd638d9a5c
4 changed files with 679 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/qgis/qgis_pool.py
+324
View File
@@ -0,0 +1,324 @@
"""
QGIS Worker 进程池管理器 — 主机侧运行。
管理 N 个长驻 Docker Worker 进程，提供任务提交和结果回收。
架构:
主机 FastAPI ←→ qgis_pool (本模块) ←→ docker exec -i ←→ qgis_worker.py (容器内)
用法:
from app.services.qgis.qgis_pool import qgis_pool
# 提交任务（阻塞等待完成）
results, summary = qgis_pool.submit_job(config, models, progress_callback)
# 关闭池
qgis_pool.shutdown()
"""
import json
import subprocess
import threading
import time
from typing import Callable, Dict, List, Optional, Tuple
from app.config.paths import get_logger
logger = get_logger("qgis.pool")
class _WorkerHandle:
"""单个 Worker 进程的句柄"""
def __init__(self, worker_id: int, proc: subprocess.Popen):
self.worker_id = worker_id
self.proc = proc
self.busy = False
self.lock = threading.Lock()
self._reader_thread: Optional[threading.Thread] = None
def is_alive(self) -> bool:
return self.proc.poll() is None
def mark_busy(self):
with self.lock:
self.busy = True
def mark_free(self):
with self.lock:
self.busy = False
def is_free(self) -> bool:
with self.lock:
return not self.busy and self.is_alive()
class QgisPool:
"""QGIS Worker 进程池"""
def __init__(self, pool_size: int = None):
"""
Args:
pool_size: 池中 Worker 数量，默认从配置读取
"""
if pool_size is None:
try:
from config import settings
pool_size = getattr(settings, "QGIS_POOL_SIZE", 2)
except Exception:
pool_size = 2
self._pool_size = pool_size
self._workers: List[_WorkerHandle] = []
self._lock = threading.Lock()
self._started = False
def _ensure_started(self):
"""懒启动：首次使用时启动 Worker 进程"""
if self._started:
return
with self._lock:
if self._started:
return
self._start_workers()
self._started = True
def _start_workers(self):
"""启动所有 Worker 进程"""
from app.services.qgis.qgis_env import (
get_docker_container,
get_docker_project_dir,
get_container_python_path,
)
container = get_docker_container()
project_dir = get_docker_project_dir()
python_path = get_container_python_path()
worker_script = f"{project_dir}/app/services/qgis/qgis_worker.py"
# Qt 平台设置
try:
from config import settings
qt_platform = getattr(settings, "QGIS_DOCKER_QT_PLATFORM", "offscreen")
except Exception:
qt_platform = "offscreen"
for i in range(self._pool_size):
cmd = [
"docker", "exec", "-i",
"-w", project_dir,
"-e", f"QT_QPA_PLATFORM={qt_platform}",
container,
python_path, worker_script,
]
logger.info(f"[Pool] 启动 Worker-{i}: {' '.join(cmd)}")
proc = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
worker = _WorkerHandle(i, proc)
self._workers.append(worker)
# 后台线程消费 stderr，防止管道满阻塞
t = threading.Thread(
target=self._drain_stderr, args=(worker,), daemon=True
)
t.start()
logger.info(f"[Pool] 已启动 {self._pool_size} 个 Worker")
@staticmethod
def _drain_stderr(worker: _WorkerHandle):
"""持续读取 Worker stderr 防止管道缓冲区满"""
try:
for line in worker.proc.stderr:
line = line.strip()
if line:
logger.debug(f"[Worker-{worker.worker_id}] stderr: {line}")
except Exception:
pass
def _find_free_worker(self, timeout: float = 30) -> Optional[_WorkerHandle]:
"""找到一个空闲 Worker，超时返回 None"""
deadline = time.time() + timeout
while time.time() < deadline:
with self._lock:
for w in self._workers:
if w.is_free():
return w
time.sleep(0.1)
return None
def submit_job(
self,
config: dict,
models: list,
progress_callback: Callable[[dict], None] = None,
) -> Tuple[list, dict]:
"""
提交一个批次任务，阻塞等待完成。
Args:
config: QGIS 配置字典
models: 模型参数列表
progress_callback: 每张图完成时的回调（接收 PROGRESS dict）
Returns:
(results_list, summary_dict)
"""
self._ensure_started()
worker = self._find_free_worker(timeout=60)
if worker is None:
raise RuntimeError("[Pool] 无可用 Worker（等待 60s 超时）")
worker.mark_busy()
try:
return self._run_job(worker, config, models, progress_callback)
finally:
# 检查 Worker 是否还活着，死了就重启
if not worker.is_alive():
logger.error(f"[Pool] Worker-{worker.worker_id} 已死亡，重启...")
self._restart_worker(worker)
worker.mark_free()
def _run_job(
self,
worker: _WorkerHandle,
config: dict,
models: list,
progress_callback: Callable[[dict], None] = None,
) -> Tuple[list, dict]:
"""向指定 Worker 发送任务并读取结果"""
job = {"config": config, "models": models}
job_json = json.dumps(job, ensure_ascii=False) + "\n"
# 写入 stdin
try:
worker.proc.stdin.write(job_json)
worker.proc.stdin.flush()
except (BrokenPipeError, OSError) as e:
raise RuntimeError(f"[Pool] 写入 Worker-{worker.worker_id} stdin 失败: {e}")
# 读取输出（直到 JOB_DONE 或 FATAL）
results = []
summary = {}
while True:
line = worker.proc.stdout.readline()
if not line:
# stdout 关闭，Worker 可能已死亡
logger.error(f"[Pool] Worker-{worker.worker_id} stdout 关闭")
summary = {"error": "Worker stdout 关闭"}
break
line = line.strip()
if not line:
continue
if line.startswith("PROGRESS:"):
try:
r = json.loads(line[len("PROGRESS:"):])
results.append(r)
if progress_callback:
progress_callback(r)
except json.JSONDecodeError:
logger.warning(f"[Pool] Worker-{worker.worker_id} PROGRESS 解析失败: {line[:200]}")
elif line.startswith("JOB_DONE:"):
try:
summary = json.loads(line[len("JOB_DONE:"):])
except json.JSONDecodeError:
summary = {"error": f"JOB_DONE 解析失败: {line[:200]}"}
break
elif line.startswith("FATAL:"):
try:
summary = json.loads(line[len("FATAL:"):])
except json.JSONDecodeError:
summary = {"error": f"FATAL: {line[:200]}"}
logger.error(f"[Pool] Worker-{worker.worker_id} FATAL: {summary}")
break
else:
logger.debug(f"[Pool] Worker-{worker.worker_id} 未知输出: {line[:200]}")
return results, summary
def _restart_worker(self, dead_worker: _WorkerHandle):
"""重启一个死亡的 Worker"""
from app.services.qgis.qgis_env import (
get_docker_container,
get_docker_project_dir,
get_container_python_path,
)
container = get_docker_container()
project_dir = get_docker_project_dir()
python_path = get_container_python_path()
worker_script = f"{project_dir}/app/services/qgis/qgis_worker.py"
try:
from config import settings
qt_platform = getattr(settings, "QGIS_DOCKER_QT_PLATFORM", "offscreen")
except Exception:
qt_platform = "offscreen"
cmd = [
"docker", "exec", "-i",
"-w", project_dir,
"-e", f"QT_QPA_PLATFORM={qt_platform}",
container,
python_path, worker_script,
]
proc = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
dead_worker.proc = proc
dead_worker.busy = False
# 重启 stderr 消费线程
t = threading.Thread(
target=self._drain_stderr, args=(dead_worker,), daemon=True
)
t.start()
logger.info(f"[Pool] Worker-{dead_worker.worker_id} 已重启")
def shutdown(self):
"""关闭所有 Worker"""
with self._lock:
for w in self._workers:
if w.proc.poll() is None:
try:
w.proc.stdin.close()
except Exception:
pass
try:
w.proc.terminate()
w.proc.wait(timeout=5)
except Exception:
try:
w.proc.kill()
except Exception:
pass
self._workers.clear()
self._started = False
logger.info("[Pool] 已关闭")
# 全局单例
qgis_pool = QgisPool()
app/services/qgis/qgis_worker.py
+191
View File
@@ -0,0 +1,191 @@
#!/usr/bin/env python3
"""
QGIS 长驻 Worker — Docker 容器内运行，常驻进程模式。
与 qgis_runner.py 的区别：
- qgis_runner.py: 一次性进程，处理完一批任务后退出
- qgis_worker.py: 长驻进程，初始化 QGIS 一次，循环从 stdin 读取任务处理
通信协议（stdin/stdout JSON Lines）：
输入: 每行一个 JSON 任务 {"config": {...}, "models": [{...}, ...]}
输出: 每张图一行 PROGRESS PROGRESS:{"name": "...", "output": "..."}
任务结束一行 JOB_DONE:{"total": N, "ok": M, "fail": K, "elapsed": T}
致命错误 FATAL:{"error": "..."}
用法:
docker exec -i qgis-server python3 /app/app/services/qgis/qgis_worker.py
"""
import json
import os
import sys
import time
# ============================================================
# 环境初始化（与 qgis_runner.py 保持一致）
# ============================================================
def _setup_python_path():
"""将项目根目录和 QGIS dist-packages 加入 sys.path"""
project_root = os.path.dirname(
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
)
if project_root not in sys.path:
sys.path.insert(0, project_root)
try:
from config import settings
pythonpath = getattr(settings, "QGIS_DOCKER_PYTHONPATH", None) or []
except Exception:
pythonpath = []
for p in pythonpath:
if os.path.isdir(p) and p not in sys.path:
sys.path.insert(0, p)
def _setup_environment():
"""设置 QGIS 运行所需的环境变量"""
os.environ["PYTHONUTF8"] = "1"
os.environ["GDAL_FILENAME_IS_UTF8"] = "YES"
os.environ["VSI_CACHE"] = "TRUE"
os.environ["VSI_CACHE_SIZE"] = "1000000"
def _init_qgis():
"""初始化 QgsApplication"""
from qgis.core import QgsApplication
try:
from config import settings
prefix_path = getattr(settings, "QGIS_DOCKER_PREFIX_PATH", "/usr")
except Exception:
prefix_path = "/usr"
QgsApplication.setPrefixPath(prefix_path, True)
qgs_app = QgsApplication([], False)
qgs_app.initQgis()
return qgs_app
# ============================================================
# 单任务处理
# ============================================================
def _emit(line: str):
"""输出一行到 stdout"""
print(line, flush=True)
def _emit_progress(result: dict):
"""输出 PROGRESS 行"""
_emit(f"PROGRESS:{json.dumps(result, ensure_ascii=False)}")
def _process_job(service, job: dict) -> dict:
"""
处理一个批次任务（多个模板），返回汇总结果。
每张图输出一行 PROGRESS，最后返回汇总 dict。
"""
models = job.get("models") or []
if not models:
return {"total": 0, "ok": 0, "fail": 0, "elapsed": 0}
ok = 0
fail = 0
t_start = time.time()
for i, model in enumerate(models):
t_model = time.time()
try:
name = service.generate(model)
result = {"name": name, "output": model["outFile"]}
_emit_progress(result)
ok += 1
elapsed = time.time() - t_model
print(
f"[worker] [{i+1}/{len(models)}] 完成: {name}, 耗时 {elapsed:.1f}s",
file=sys.stderr,
)
except Exception as e:
elapsed = time.time() - t_model
error_msg = f"{e}"
fail_result = {"name": model.get("name", ""), "output": "", "error": error_msg}
_emit_progress(fail_result)
fail += 1
print(
f"[worker] [{i+1}/{len(models)}] 失败: {model.get('name', '?')}, "
f"耗时 {elapsed:.1f}s — {error_msg}",
file=sys.stderr,
)
total_elapsed = time.time() - t_start
summary = {"total": len(models), "ok": ok, "fail": fail, "elapsed": round(total_elapsed, 1)}
print(
f"[worker] 批次完成: {ok}成功/{fail}失败, 总耗时 {total_elapsed:.1f}s",
file=sys.stderr,
)
return summary
# ============================================================
# 主循环
# ============================================================
def main():
_setup_environment()
_setup_python_path()
print("[worker] 正在初始化 QGIS...", file=sys.stderr)
t_init = time.time()
qgs_app = _init_qgis()
print(f"[worker] QGIS 初始化完成 ({time.time() - t_init:.1f}s)，等待任务...", file=sys.stderr)
try:
from app.services.qgis.map_service import MapService
# 持有 MapService 实例（config 在第一个任务时设置）
service = None
for line_no, line in enumerate(sys.stdin, 1):
line = line.strip()
if not line:
continue
try:
job = json.loads(line)
except json.JSONDecodeError as e:
print(f"[worker] 第{line_no}行 JSON 解析失败: {e}", file=sys.stderr)
_emit(f'JOB_DONE:{{"error": "JSON解析失败: {e}"}}')
continue
config = job.get("config")
if config:
# 更新 config（每个任务可以带不同的 config）
service = MapService(config)
if service is None:
_emit(f'JOB_DONE:{{"error": "未提供 config"}}')
continue
summary = _process_job(service, job)
_emit(f"JOB_DONE:{json.dumps(summary, ensure_ascii=False)}")
# 清空项目，恢复到初始状态，等待下一个任务
from qgis.core import QgsProject
QgsProject.instance().clear()
print("[worker] 项目已清空，等待下一个任务", file=sys.stderr)
except KeyboardInterrupt:
print("[worker] 收到中断信号，退出", file=sys.stderr)
except Exception as e:
print(f"[worker] 致命错误: {e}", file=sys.stderr)
_emit(f'FATAL:{{"error": "{e}"}}')
finally:
qgs_app.exitQgis()
print("[worker] QGIS 已退出", file=sys.stderr)
if __name__ == "__main__":
main()