mark.tian/ai-talk-callback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

不处理退出信号

Browse Source
This commit is contained in:
liangtianyu
2025-12-14 20:20:30 +08:00
parent 8541dc77ec
commit 25a016977c
1 changed files with 71 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

120
main.py
View File

@@ -1,4 +1,3 @@
from datetime import datetime
import os
import traceback
@@ -22,6 +21,7 @@ from sqlalchemy import text
LoggerManager.setup_logging()
logger = get_main_logger()
def global_exception_handler(exc_type, exc_value, exc_traceback):
if issubclass(exc_type, KeyboardInterrupt):
sys.__excepthook__(exc_type, exc_value, exc_traceback)
@@ -32,7 +32,7 @@ def global_exception_handler(exc_type, exc_value, exc_traceback):
error_msg += f"异常类型: {exc_type.__name__}\n"
error_msg += f"异常信息: {exc_value}\n"
error_msg += "堆栈跟踪：\n"
error_msg += ''.join(traceback.format_tb(exc_traceback))
error_msg += "".join(traceback.format_tb(exc_traceback))
error_msg += f"异常时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
error_msg += "-" * 50 + "\n"
@@ -42,14 +42,16 @@ def global_exception_handler(exc_type, exc_value, exc_traceback):
# 输出到文件
logger.error(error_msg)
# 注册全局异常处理器
sys.excepthook = global_exception_handler
@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动时初始化
logger.info("🚀 应用启动中...")
# Redis 连接对象
redis_client = None
@@ -76,10 +78,10 @@ async def lifespan(app: FastAPI):
# 执行 ping 命令验证连接
await redis_client.ping()
logger.info("✅ Redis 连接验证成功")
# 存储到应用状态中供其他组件使用
app.state.redis_client = redis_client
except Exception as redis_error:
logger.error(f"❌ Redis 连接验证失败: {redis_error}")
raise
@@ -96,7 +98,7 @@ async def lifespan(app: FastAPI):
finally:
# 关闭时清理
logger.info("🛑 应用关闭中...")
# 关闭 Redis 连接
if redis_client:
try:
@@ -104,7 +106,7 @@ async def lifespan(app: FastAPI):
logger.info("🔴 Redis 连接已关闭")
except Exception as e:
logger.warning(f"⚠️ 关闭 Redis 连接时出现警告: {e}")
logger.info("👋 应用已关闭")
@@ -161,17 +163,18 @@ async def health_check():
raise HTTPException(status_code=404, detail="Not Found")
logger.debug("💓 健康检查接口被访问")
return {"status": "healthy"}
# 全局变量存储进程
processes = []
def signal_handler(signum, frame):
"""信号处理器，用于优雅关闭所有服务"""
print(f"\n🛑 接收到信号 {signum}，正在关闭所有服务...")
# 逆序关闭进程（最后启动的最先关闭）
for i, process in enumerate(reversed(processes)):
if process and process.poll() is None: # 进程仍在运行
@@ -185,86 +188,104 @@ def signal_handler(signum, frame):
process.kill() # 强制杀死进程
except Exception as e:
print(f"❌ 关闭进程时出错: {e}")
print("👋 所有服务已关闭")
sys.exit(0)
def start_all_services():
"""启动所有服务"""
global processes
print("\n🚀 AI Talk Callback API 一键启动所有服务")
print("=" * 60)
# 注册信号处理器
signal.signal(signal.SIGINT, signal_handler) # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler) # 终止信号
# signal.signal(signal.SIGINT, signal_handler) # Ctrl+C
# signal.signal(signal.SIGTERM, signal_handler) # 终止信号
try:
# 1. 启动 Celery Worker
print("🌿 启动 Celery Worker...")
worker_process = subprocess.Popen([
sys.executable, "-m", "celery",
"-A", "app.celery_app",
"worker",
'--loglevel=info',
'--pool=solo',
'--concurrency=1',
'--time-limit=300', # 5分钟任务超时
'--soft-time-limit=240' # 4分钟软超时
])
worker_process = subprocess.Popen(
[
sys.executable,
"-m",
"celery",
"-A",
"app.celery_app",
"worker",
"--loglevel=info",
"--pool=solo",
"--concurrency=1",
"--time-limit=300", # 5分钟任务超时
"--soft-time-limit=240", # 4分钟软超时
]
)
processes.append(worker_process)
time.sleep(2) # 等待 Worker 启动
# 2. 启动 Celery Beat
print("\n📅 启动 Celery Beat...")
beat_process = subprocess.Popen([
sys.executable, "-m", "celery",
"-A", "app.celery_app",
"beat",
'--loglevel=info',
f'--schedule={os.path.join(tempfile.gettempdir(), "celerybeat-schedule")}'
])
beat_process = subprocess.Popen(
[
sys.executable,
"-m",
"celery",
"-A",
"app.celery_app",
"beat",
"--loglevel=info",
f'--schedule={os.path.join(tempfile.gettempdir(), "celerybeat-schedule")}',
]
)
processes.append(beat_process)
time.sleep(2) # 等待 Beat 启动
# 3. 启动 Flower 监控（如果启用）
if settings.flower_enabled:
print("\n📊 启动 Flower 监控服务...")
flower_cmd = [
sys.executable, "-m", "celery",
"-A", "app.celery_app",
sys.executable,
"-m",
"celery",
"-A",
"app.celery_app",
f"--broker={settings.celery_broker_url}",
"flower",
f"--port={settings.flower_port}"
f"--port={settings.flower_port}",
]
if settings.flower_basic_auth:
flower_cmd.append(f"--basic_auth={settings.flower_basic_auth}")
if settings.flower_url_prefix:
flower_cmd.append(f"--url_prefix={settings.flower_url_prefix}")
flower_process = subprocess.Popen(flower_cmd)
processes.append(flower_process)
time.sleep(2) # 等待 Flower 启动
# 4. 启动 FastAPI 应用
print("\n🚀 启动 FastAPI 应用...")
api_cmd = [
sys.executable, "-m", "uvicorn",
sys.executable,
"-m",
"uvicorn",
"main:app",
"--host", "0.0.0.0",
"--port", "8000"
"--host",
"0.0.0.0",
"--port",
"8000",
]
# 添加调试模式（如果配置了）
if settings.debug:
api_cmd.append("--reload")
api_process = subprocess.Popen(api_cmd)
processes.append(api_process)
time.sleep(2) # 等待 API 启动
print("\n" + "=" * 60)
print("✅ 所有服务启动完成！")
print("\n🌐 服务地址:")
@@ -275,7 +296,7 @@ def start_all_services():
print(f" 📊 监控界面: {settings.flower_url}")
print("\n💡 使用 Ctrl+C 可以优雅关闭所有服务")
print("=" * 60)
# 等待所有进程
while True:
# 检查是否有进程异常退出
@@ -284,15 +305,16 @@ def start_all_services():
print(f"❌ 进程 {i+1} 异常退出，退出码: {process.returncode}")
signal_handler(signal.SIGINT, None)
return
time.sleep(1) # 每秒检查一次
except KeyboardInterrupt:
signal_handler(signal.SIGINT, None)
except Exception as e:
print(f"❌ 启动服务时出错: {e}")
signal_handler(signal.SIGINT, None)
if __name__ == "__main__":
# 直接启动所有服务
start_all_services()