cjw/geMoldInsight
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加celery任务队列，任务持久化（服务重启不丢）、支持多 Worker 水平扩展、失败自动重试

Browse Source
This commit is contained in:
chenjw28
2026-05-28 17:59:39 +08:00
parent 6ad22e4466
commit 44902207c9
9 changed files with 133 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/settings.py
+2
View File
@@ -66,6 +66,8 @@ class Settings:
self.ENABLE_FREECAD_VERIFICATION = os.getenv("ENABLE_FREECAD_VERIFICATION", "false").lower() == "true"
self.FREECAD_VERIFICATION_TIMEOUT = int(os.getenv("FREECAD_VERIFICATION_TIMEOUT", "120"))
self.PROCESSING_TIMEOUT_BASE = int(os.getenv("PROCESSING_TIMEOUT_BASE", "300"))
self.PROCESSING_TIMEOUT_PER_MB = int(os.getenv("PROCESSING_TIMEOUT_PER_MB", "15"))
# Redis
self.REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
docker-compose.yml
+28
View File
@@ -80,6 +80,34 @@ services:
networks:
- moldinsight_network
celery-worker:
build: .
container_name: moldinsight_celery
command: >
/bin/bash -c "source /opt/conda/etc/profile.d/conda.sh && conda activate moldinsight && cd /app/src && celery -A celery_app worker --concurrency=2 --loglevel=info"
volumes:
- ./uploads:/app/uploads
- ./html_output:/app/html_output
- ./logs:/app/logs
- ./.env:/app/.env:ro
environment:
- DB_HOST=postgres
- DB_PORT=5432
- DB_NAME=${DB_NAME:-moldinsight}
- DB_USER=${DB_USER:-moldinsight_user}
- DB_PASSWORD=${DB_PASSWORD:-moldinsight_password}
- RUSTFS_ENDPOINT=http://minio:9000
- RUSTFS_ACCESS_KEY=${MINIO_ACCESS_KEY:-minioadmin}
- RUSTFS_SECRET_KEY=${MINIO_SECRET_KEY:-minioadmin}
depends_on:
postgres:
condition: service_healthy
minio:
condition: service_healthy
restart: unless-stopped
networks:
- moldinsight_network
volumes:
postgres_data:
driver: local
requirements.txt
+1
View File
@@ -44,6 +44,7 @@ aiohttp>=3.13.4
# ============================================
# 消息队列
# ============================================
celery[redis]>=5.3.0
kafka-python>=2.0.2
redis>=4.5.0
src/api/routes.py
+3 -10
View File
@@ -1,5 +1,5 @@
# api/routes.py
from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks, Request, Depends
from fastapi import APIRouter, UploadFile, File, HTTPException, Request, Depends
from typing import Optional, Dict, Any, List
import uuid
from datetime import datetime
@@ -10,8 +10,8 @@ from models.schemas import ProcessingStatus, create_task_info
from utils.file_handler import FileHandler
from services.storage_integration_rustfs import StorageIntegrationService
from services.redis_task_manager import redis_task_manager
from services.processing_service import processing_service
from services.task_query_service import TaskQueryService
from celery_tasks import process_stp_task
from database.database import get_db_session
from utils.logger import get_logger
from sqlalchemy.ext.asyncio import AsyncSession
@@ -54,7 +54,6 @@ async def health():
@router.post("/upload")
async def upload_stp(
background_tasks: BackgroundTasks,
file: UploadFile = File(...),
material: Optional[str] = "ABS",
db_session: AsyncSession = Depends(get_db_session),
@@ -104,13 +103,7 @@ async def upload_stp(
# 后台处理走统一编排服务，避免请求会话在后台失效
process_params = {"material": material}
background_tasks.add_task(
processing_service.process_file_with_storage,
task_id,
file_path,
stp_file.id,
process_params,
)
process_stp_task.delay(task_id, str(file_path), stp_file.id, process_params)
logger.info(f"[UPLOAD] 后台处理已调度: task_id={task_id}")
return {
src/api/v1/upload_router.py
+3 -7
View File
@@ -1,5 +1,5 @@
# api/v1/upload_router.py
from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks, Depends, Form
from fastapi import APIRouter, UploadFile, File, HTTPException, Depends, Form
import uuid
from datetime import datetime
from pathlib import Path
@@ -12,8 +12,8 @@ from database.database import get_db_session
from utils.logger import get_logger
from sqlalchemy.ext.asyncio import AsyncSession
from services.auth_service import get_current_active_user
from services.processing_service import processing_service
from models.database import User
from celery_tasks import process_stp_task
logger = get_logger(__name__)
@@ -24,7 +24,6 @@ file_handler = FileHandler()
@router.post("/upload")
async def upload_stp(
background_tasks: BackgroundTasks,
file: UploadFile = File(...),
material: str = Form(...),
draft_angle: float = Form(...),
@@ -91,10 +90,7 @@ async def upload_stp(
task_info["file_hash"] = file_meta["sha256"]
await redis_task_manager.set_task(task_id, task_info)
background_tasks.add_task(
processing_service.process_file_with_storage,
task_id, file_path, stp_file.id, process_params
)
process_stp_task.delay(task_id, str(file_path), stp_file.id, process_params)
logger.info(f"[UPLOAD] 后台处理已调度: task_id={task_id}")
return {
src/celery_app.py
+34
View File
@@ -0,0 +1,34 @@
import os
from dotenv import load_dotenv
from celery import Celery
load_dotenv()
redis_host = os.getenv("REDIS_HOST", "localhost")
redis_port = os.getenv("REDIS_PORT", "6379")
redis_password = os.getenv("REDIS_PASSWORD", "")
redis_db = os.getenv("REDIS_DB", "0")
if redis_password:
broker_url = f"redis://:{redis_password}@{redis_host}:{redis_port}/{redis_db}"
else:
broker_url = f"redis://{redis_host}:{redis_port}/{redis_db}"
app = Celery(
"moldinsight",
broker=broker_url,
backend=broker_url,
include=["celery_tasks"],
)
app.conf.update(
task_serializer="json",
result_serializer="json",
accept_content=["json"],
task_acks_late=True,
worker_prefetch_multiplier=1,
task_track_started=True,
task_soft_time_limit=1800,
task_time_limit=2100,
broker_connection_retry_on_startup=True,
)
src/celery_tasks.py
+27
View File
@@ -0,0 +1,27 @@
from celery_app import app
from services.processing_service import processing_service
from utils.logger import get_logger
logger = get_logger(__name__)
@app.task(bind=True, max_retries=1, default_retry_delay=60)
def process_stp_task(self, task_id: str, file_path: str, stp_file_id: int,
process_params: dict):
"""Celery 任务：异步处理 STP 文件生成模具型腔"""
import asyncio
try:
logger.info(f"[celery] 开始处理: {task_id}")
asyncio.run(processing_service.process_file_with_storage(
task_id, file_path, stp_file_id, process_params
))
logger.info(f"[celery] 处理完成: {task_id}")
return {"task_id": task_id, "status": "completed"}
except Exception as exc:
logger.error(f"[celery] 处理失败: {task_id}, error={exc}")
try:
self.retry(exc=exc)
except Exception:
pass
raise
src/core/geometry_analyzer.py
+18 -19
View File
@@ -1,5 +1,6 @@
from typing import Dict, List, Any, Optional
import math
from concurrent.futures import ThreadPoolExecutor, as_completed
import numpy as np
from OCC.Core.TopoDS import TopoDS_Shape
from models.schemas import (
@@ -74,27 +75,25 @@ class GeometryAnalyzer:
def _detect_features(self, geometry_data: Dict[str, Any],
shape: Optional[TopoDS_Shape] = None) -> List[Dict[str, Any]]:
"""检测模具特征"""
features = []
"""检测模具特征 — 独立检测并行执行"""
features: List[Dict[str, Any]] = []
wall_features = self._detect_wall_features(geometry_data, shape)
features.extend(wall_features)
with ThreadPoolExecutor(max_workers=4, thread_name_prefix="feat") as pool:
futures = {
pool.submit(self._detect_wall_features, geometry_data, shape): "wall",
pool.submit(self._detect_rib_features, geometry_data, shape): "rib",
pool.submit(self._detect_boss_features, geometry_data, shape): "boss",
pool.submit(self._analyze_draft_angles, geometry_data, shape): "draft",
}
if shape is not None:
futures[pool.submit(self._detect_curvature_features, shape)] = "curvature"
futures[pool.submit(self._detect_fillet_features, shape)] = "fillet"
rib_features = self._detect_rib_features(geometry_data, shape)
features.extend(rib_features)
boss_features = self._detect_boss_features(geometry_data, shape)
features.extend(boss_features)
draft_features = self._analyze_draft_angles(geometry_data, shape)
features.extend(draft_features)
if shape is not None:
curvature_features = self._detect_curvature_features(shape)
features.extend(curvature_features)
fillet_features = self._detect_fillet_features(shape)
features.extend(fillet_features)
for future in as_completed(futures):
try:
features.extend(future.result())
except Exception:
pass
logger.info(f"检测到 {len(features)} 个特征")
return features
src/services/processing_service.py
+17 -11
View File
@@ -65,8 +65,15 @@ class ProcessingService:
try:
logger.info(f"开始处理文件并生成模具型腔: {file_path}")
# 设置处理超时（5分钟）
timeout_seconds = 300
from config.settings import settings
file_size_bytes = Path(file_path).stat().st_size if Path(file_path).exists() else 0
file_size_mb = max(file_size_bytes / (1024 * 1024), 1)
timeout_seconds = min(
max(settings.PROCESSING_TIMEOUT_BASE, int(file_size_mb * settings.PROCESSING_TIMEOUT_PER_MB)),
1800,
)
logger.info(f"处理超时设置为 {timeout_seconds}s (文件 {file_size_mb:.1f}MB)")
try:
await asyncio.wait_for(
@@ -575,7 +582,7 @@ class ProcessingService:
pointcloud_data: Optional[Dict[str, Any]] = None,
lod_data: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""为每个候选分模方案生成独立HTML预览链接。"""
"""为候选分模方案生成轻量摘要链接（完整HTML仅最优方案按需生成）"""
candidate_schemes = detailed_cavity_json.get("candidate_schemes", [])
if not candidate_schemes:
return detailed_cavity_json
@@ -585,15 +592,14 @@ class ProcessingService:
if not cavity_data:
continue
suffix = scheme.get("scheme_id")
html_path = self.html_generator.generate_and_save_visualization(
geometry_data,
stp_filename,
cavity_data=cavity_data,
pointcloud_data=pointcloud_data,
suffix=suffix,
lod_data=lod_data,
base_stem = Path(stp_filename).stem.replace(" ", "_")
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
summary_name = f"mold_{base_stem}_{suffix}_{ts}_summary.json"
summary_content = self.html_generator.generate_3d_viewer_summary(
geometry_data, cavity_data
)
scheme["html_file"] = f"/html/{Path(html_path).name}"
self.html_generator.save_data_file(summary_content, summary_name)
scheme["summary_file"] = f"/html/{summary_name}"
best_scheme = CalculationService.get_best_scheme(detailed_cavity_json)
if best_scheme: