"机器人跑着跑着突然不动了"——这种情况十有八九不是代码挂了，是微信连接断了，但你的程序不知道。等你发现已经过了好几个小时，客户消息堆积成山。解决这个问题需要两件事：周期性健康检测和连接异常时的分级处理。

一、心跳检测：两种方式选哪种

方式一：主动调 /account/status 接口，平台返回你账号当前的在线状态、最后一次消息收发时间、当前设备标识。这个接口是平台侧封装的，准确但有频率限制——一般每分钟最多调一次，别高频轮询。

方式二：被动观察消息回调的间隔，比如正常情况下每小时至少收到几条消息，如果连续 30 分钟没有任何回调进来，就判定连接有问题。这种方式不需要额外调接口，但不够灵敏——如果刚好那段时间没人发消息，就会误判。

实际用的时候两种都要：主动心跳每 30 秒调一次 /account/status，配合被动观察回调间隔做交叉验证。两个信号同时报警才算真的断了，单独一个可能是误报。

二、连接异常分三级

级别	现象	处理
轻微	消息延迟 5-10 秒、偶尔丢 1 条	等 2 分钟自己恢复，记录日志
中等	连续 3 次心跳返回 offline	触发重连接口 /account/reconnect，1 分钟后再检
严重	重连 3 次都失败、回调中断超过 10 分钟	通知运维、标记账号为"不可用"、流量切到备用号

中等和严重的分界线很重要——重连是个昂贵动作，连续重连反而可能加重服务器压力。设一个重连冷却时间（比如 5 分钟内只重连一次），别让重连风暴把自己打死。

三、重连后的消息补发

连接恢复之后，回调会补发断开期间积压的消息吗？大部分平台会，但不是所有——有些平台只补发最近 5 分钟的，超过的就丢了。所以重连之后要做两件事：

1. 主动调 /message/history 拉取断开期间的历史消息，确保没漏掉

2. 拉取最新的好友列表和群成员列表，因为断开期间可能有人进群或退群

账号状态管理链路对照

环节	做法	关键点	常见坑
心跳	30s 主动 + 回调间隔被动	交叉验证防误报	只做一种不够
分级	延迟/重连/切备用	重连 5 分钟冷却	重连风暴打死自己
补发	主动拉历史+刷新列表	平台补发可能不全	等平台补发就丢消息
恢复	标记账号可用+重置计数	写监控告警	恢复了不通知运维

健康检测与重连实现

import time, threading, collections, datetime

class AccountHealthMonitor:
    HEARTBEAT_INTERVAL = 30
    PASSIVE_TIMEOUT = 1800        # 30 分钟无回调 → 异常
    MAX_RETRY_FORCE_RECONNECT = 3
    RECONNECT_COOLDOWN = 300      # 重连冷却 5 分钟
    RECOVERY_FETCH_WINDOW = 900   # 重连后拉 15 分钟历史

    def __init__(self, api, store):
        self.api = api
        self.store = store
        self._last_callback_at = time.time()
        self._reconnect_lock = threading.Lock()
        self._last_reconnect_at = 0
        threading.Thread(target=self._heartbeat_loop, daemon=True).start()
        threading.Thread(target=self._passive_loop, daemon=True).start()

    def on_any_callback(self):
        self._last_callback_at = time.time()

    def _heartbeat_loop(self):
        while True:
            try:
                status = self.api.get("/account/status").json()
                online = status.get("online", False)
                last_msg_at = status.get("last_msg_ts", 0)
                self.store.set_account_status({
                    "online": online,
                    "last_msg_at": last_msg_at,
                    "checked_at": int(time.time())})
                if not online:
                    self._handle_mid_failure("heartbeat_offline")
            except Exception as e:
                self.store.log_fail("heartbeat", str(e))
                self._handle_mid_failure("heartbeat_error")
            time.sleep(self.HEARTBEAT_INTERVAL)

    def _passive_loop(self):
        while True:
            gap = time.time() - self._last_callback_at
            if gap > self.PASSIVE_TIMEOUT:
                # 被动信号：超过 30 分钟无任何回调
                self.store.log_warn("passive_timeout", f"gap={gap}s")
                # 被动信号只做日志，触发重连需要主动信号也异常
            time.sleep(60)

    def _handle_mid_failure(self, reason):
        # 中等故障：触发重连
        now = time.time()
        if now - self._last_reconnect_at < self.RECONNECT_COOLDOWN:
            self.store.log_warn("reconnect_cooldown", reason)
            return
        if not self._reconnect_lock.acquire(blocking=False):
            return
        try:
            self._last_reconnect_at = now
            for i in range(self.MAX_RETRY_FORCE_RECONNECT):
                try:
                    self.api.post("/account/reconnect").json()
                    self._post_recovery()
                    self.store.set_account_status(
                        {"online": True, "recovered": True})
                    return
                except Exception as e:
                    self.store.log_fail("reconnect", f"attempt={i+1} {e}")
                    time.sleep(2 ** i)
            # 严重故障：标记不可用+通知运维+切备用
            self.store.alert("account_unavailable", reason)
            self.store.mark_account_disabled()
            self._switch_to_backup()
        finally:
            self._reconnect_lock.release()

    def _post_recovery(self):
        # 重连后：主动拉历史+刷列表
        since = int(time.time()) - self.RECOVERY_FETCH_WINDOW
        history = self.api.get("/message/history",
                               params={"since": since}).json()["items"]
        # 交给正常的消息处理逻辑
        for m in history:
            self.store.enqueue_for_processing(m)
        # 刷好友和群列表
        self.api.get("/friend/list").json()
        self.api.get("/group/list").json()

    def _switch_to_backup(self):
        accounts = self.store.all_accounts()
        for acc in accounts:
            if acc["status"] == "active" and acc["id"] != self.store.current_account_id():
                self.store.promote_backup(acc["id"])
                self.api.set_active_account(acc["id"])
                break

落地建议

账号健康检测靠主动心跳 + 被动回调间隔交叉验证，两种信号同时报警才触发重连，单独一个可能是误报。重连有冷却时间（比如 5 分钟内只允许一次），别让重连风暴把自己打死；连续重连 3 次还不行就升级到"严重故障"，标记账号不可用、切备用号、通知运维。重连之后要主动拉取断开期间的历史消息（别等平台补发），同时刷新好友和群列表——断开期间可能有人进群退群。/account/status 和 /account/reconnect 这两个能力由 Eyun 这类平台提供，分级策略、冷却时间、切备用的逻辑在你自己的服务里实现。