Python应用中atexit失效致用户离线状态更新失败的解决方案咨询

atexit仅能处理脚本正常结束、主动调用sys.exit()这类场景，对于断电、断网、强制杀进程等异常终止情况，应用根本没有机会执行离线状态更新代码。核心解决思路是从「主动上报离线」转为「被动检测在线状态」，以下是具体可行方案：

方案1：新增「最后活跃时间」字段，通过超时判断状态 #

这是最可靠的方案，完全不依赖应用主动触发离线更新，而是通过数据库记录的活跃痕迹判断用户状态。

步骤1：修改数据库表结构 #

新增last_active字段，用于记录用户最后一次与数据库交互的时间：

ALTER TABLE bot_accounts ADD COLUMN last_active TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP;

该字段会在每次更新该行时自动刷新为当前时间，也可以手动触发更新。

步骤2：修改应用代码 #

添加后台心跳线程持续刷新活跃时间，同时保留正常退出时的离线更新逻辑，还可以捕获常见终止信号覆盖更多场景：

import mysql.connector
import atexit
import threading
import time
import signal

db_connection = mysql.connector.connect(
  host="********",
  user="********",
  password="********",
  database="********"
)
serial_key = "用户的serial_key"  # 替换为实际的用户标识
HEARTBEAT_INTERVAL = 30  # 心跳间隔（秒）
ONLINE_THRESHOLD = 300  # 离线判断阈值（5分钟=300秒）

def update_active_status():
    """更新在线状态与最后活跃时间，使用参数化查询避免SQL注入"""
    try:
        if db_connection.is_connected():
            cursor = db_connection.cursor()
            cursor.execute(
                "UPDATE bot_accounts SET status = 'online', last_active = NOW() WHERE serial_key = %s ORDER BY id ASC LIMIT 1;",
                (serial_key,)
            )
            db_connection.commit()
            cursor.close()
    except Exception as e:
        print(f"心跳更新失败: {str(e)}")

def heartbeat_loop():
    """后台心跳线程，持续刷新活跃时间"""
    while True:
        update_active_status()
        time.sleep(HEARTBEAT_INTERVAL)

def set_offline():
    """正常退出或捕获信号时，主动设置离线状态"""
    try:
        if db_connection.is_connected():
            cursor = db_connection.cursor()
            cursor.execute(
                "UPDATE bot_accounts SET status = 'offline' WHERE serial_key = %s ORDER BY id ASC LIMIT 1;",
                (serial_key,)
            )
            db_connection.commit()
            cursor.close()
            print("用户状态已设为离线")
    except Exception as e:
        print(f"设置离线失败: {str(e)}")

# 捕获强制终止信号（如Ctrl+C、系统杀进程）
def handle_terminate_signal(signum, frame):
    set_offline()
    exit(0)

signal.signal(signal.SIGINT, handle_terminate_signal)
signal.signal(signal.SIGTERM, handle_terminate_signal)

# 启动守护线程执行心跳
heartbeat_thread = threading.Thread(target=heartbeat_loop, daemon=True)
heartbeat_thread.start()

# 初始化在线状态
update_active_status()
print("用户状态已设为在线")

# 注册正常退出的离线处理
atexit.register(set_offline)

# 主应用逻辑示例
while True:
    # 你的业务代码，比如处理用户请求、任务执行等
    time.sleep(1)

步骤3：查询用户真实在线状态 #

通过last_active与当前时间的差值判断用户是否在线，无需依赖status字段的实时更新：

SELECT 
    serial_key,
    CASE 
        WHEN last_active > NOW() - INTERVAL 300 SECOND THEN 'online'
        ELSE 'offline'
    END AS current_status
FROM bot_accounts
WHERE serial_key = '目标serial_key';

关键注意事项 #

• 杜绝SQL注入：绝对不能用字符串拼接SQL，必须使用参数化查询（代码中的%s占位符），防止恶意输入攻击。
• 守护线程设置：心跳线程设为守护线程，确保主进程退出时线程自动终止，避免残留僵尸进程。
• 异常捕获：数据库操作必须添加异常处理，防止连接断开导致心跳线程崩溃。

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