如何在Gunicorn部署的Flask应用中正确处理SIGTERM信号？

我有一个Python Flask应用，代码如下：

from flask import Flask, Response
import signal

app = Flask("myapp")

def exit_gracefully(*args):
    print("Doing something here")

signal.signal(signal.SIGINT, exit_gracefully)
signal.signal(signal.SIGTERM, exit_gracefully)

@app.route("/myendpoint")
def myendpoint():
    print("Doing something here")
    return Response("Everything is fine", status=200, mimetype="application/json")

该应用通过Gunicorn部署，依赖标准wsgi.py文件和systemd服务：

systemd配置文件：

[Unit]
Description=myservice

[Service]
User=root
Group=root
WorkingDirectory=/opt/myservice
ExecStart=/opt/myservice/run.sh

[Install]
WantedBy=multi-user.target

启动脚本run.sh：

#!/bin/bash
source env/bin/activate
env/bin/gunicorn --bind unix:gunicorn.sock --workers 1 --timeout 10 wsgi:app

我需要拦截服务收到的SIGTERM信号来执行清理操作（例如设置threading.Event()标志），但执行systemctl stop myservice时，服务会挂起30秒才关闭（推测是Gunicorn默认的graceful_timeout值），无法立即停止。

查看Gunicorn源码发现，Worker类会注册自身的信号处理器实现优雅关闭：

# gunicorn/workers/base.py:173
signal.signal(signal.SIGTERM, self.handle_exit)

我推测是自定义信号处理器覆盖了Gunicorn的处理器，但我需要同时执行自定义清理逻辑和Gunicorn Worker的正常停止流程。另外，直接在shell中运行run.sh并按Ctrl+C发送SIGINT时，应用能立即停止。

请问该如何处理？我曾考虑在自定义处理器中调用handle_exit，但代码与Gunicorn解耦，不知如何实现，是否需要调整应用设计？

方案1：保存并调用原信号处理器 #

在注册自定义信号处理器前，先保存Gunicorn已注册的原生处理器，执行完自定义清理逻辑后再调用原处理器，确保Gunicorn的正常关闭流程不受影响：

from flask import Flask, Response
import signal

app = Flask("myapp")

# 保存Gunicorn注册的原生信号处理器
original_sigint = signal.getsignal(signal.SIGINT)
original_sigterm = signal.getsignal(signal.SIGTERM)

def exit_gracefully(sig, frame):
    # 执行自定义清理操作
    print("执行自定义清理逻辑...")
    # 示例：设置threading.Event()标志
    # shutdown_event.set()
    
    # 调用原处理器，触发Gunicorn的正常关闭流程
    if sig == signal.SIGINT and original_sigint:
        original_sigint(sig, frame)
    elif sig == signal.SIGTERM and original_sigterm:
        original_sigterm(sig, frame)

signal.signal(signal.SIGINT, exit_gracefully)
signal.signal(signal.SIGTERM, exit_gracefully)

@app.route("/myendpoint")
def myendpoint():
    print("处理请求中")
    return Response("Everything is fine", status=200, mimetype="application/json")

注意：此方案依赖信号处理器的注册顺序，需确保你的应用代码在Gunicorn注册Worker信号处理器之后执行（Gunicorn会在fork Worker进程后注册处理器，因此在Flask全局代码中保存的是父进程的处理器，可能需要调整为在Worker进程启动后注册）。

方案2：自定义Gunicorn Worker类（推荐） #

直接继承Gunicorn的Worker类，重写handle_exit方法，在Gunicorn的关闭流程中嵌入自定义清理逻辑，这是最可靠的方式，避免信号处理器覆盖问题：

1. 创建custom_worker.py文件：

from gunicorn.workers.sync import SyncWorker
# 如果使用异步Worker，替换为对应的类，比如gunicorn.workers.ggevent.GeventWorker

class CustomWorker(SyncWorker):
    def handle_exit(self, sig, frame):
        # 先执行自定义清理逻辑
        print("执行自定义清理操作...")
        # 示例：设置线程退出标志
        # self.app.shutdown_event.set()
        
        # 调用父类的handle_exit方法，完成Gunicorn的优雅关闭流程
        super().handle_exit(sig, frame)

2. 修改启动脚本run.sh，指定使用自定义Worker类：

#!/bin/bash
source env/bin/activate
env/bin/gunicorn --bind unix:gunicorn.sock --workers 1 --timeout 10 --worker-class=custom_worker.CustomWorker wsgi:app

方案3：调整systemd服务配置（辅助优化） #

默认情况下，systemd发送SIGTERM给服务，如果你希望改用SIGINT（你测试过SIGINT能立即停止），可以修改systemd服务文件，添加KillSignal配置：

[Service]
User=root
Group=root
WorkingDirectory=/opt/myservice
ExecStart=/opt/myservice/run.sh
KillSignal=SIGINT

此方案仅为辅助优化，建议配合方案1或2使用，确保自定义清理逻辑能被触发。

内容的提问来源于stack exchange，提问作者Ailothaen