第二代Google Cloud Functions如何处理SIGTERM信号？

第二代Cloud Functions基于Cloud Run构建，确实支持SIGTERM信号处理，但直接在全局作用域注册信号处理器的方式无法生效——因为Functions Framework会接管主进程的信号处理逻辑，覆盖自定义的处理器设置。以下是两种可行的实现方式：

方式一：使用atexit模块注册退出清理函数 #

atexit模块会在进程退出前自动执行注册的函数，适用于大多数通用清理场景（无论进程是正常退出还是因SIGTERM终止）：

import functions_framework
import atexit
import sys

def cleanup():
    print("进程即将退出，执行清理操作...", flush=True)
    # 在这里添加你的清理逻辑：关闭数据库连接、保存临时数据、释放资源等
    sys.exit(0)

# 注册清理函数
atexit.register(cleanup)

@functions_framework.http
def main(request):
    print("Hi!")
    return 'OK'

方式二：后台线程监听SIGTERM信号 #

如果需要精确响应SIGTERM信号（比如区分不同终止信号），可以通过后台线程监听信号，避免被Functions Framework的信号处理逻辑覆盖：

import functions_framework
import signal
import sys
import threading
import time

# 用于标记是否收到终止信号
shutdown_event = threading.Event()

def shutdown_handler(signal_num, frame):
    print("收到SIGTERM信号，开始执行优雅关闭逻辑...", flush=True)
    # 在这里添加你的优雅关闭逻辑
    shutdown_event.set()
    sys.exit(0)

def signal_listener():
    # 注册SIGTERM信号处理器
    signal.signal(signal.SIGTERM, shutdown_handler)
    # 保持线程运行，直到收到终止信号
    while not shutdown_event.is_set():
        time.sleep(1)

# 启动后台监听线程（设置为守护线程，随主进程退出）
threading.Thread(target=signal_listener, daemon=True).start()

@functions_framework.http
def main(request):
    print("Hi!")
    return 'OK'

注意事项 #

• Cloud Functions实例终止时有90秒的宽限期，所有清理操作必须在该时间内完成，否则进程会被强制发送SIGKILL信号终止。
• 确保清理逻辑中的输出使用flush=True，避免日志因缓冲区未刷新而丢失。

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