我维护过一个网页变更监控服务。上线初期，最常收到的使用者反馈不是“漏报了”。而是——

一个被设成免打扰的告警程序，等于没有告警程序。这篇聊的是告警这一层的设计：去重、静默、聚合、限流、重试、幂等。老实说，都是些不性感的东西，但决定了使用者会不会一直用下去。按理说，

一、先把「检测」和「通知」拆开

使用者痛点：告警策略每改一次都要改动主要检测代码；话说回来，通知失败会导致数据丢失；想回溯阈值变化时找不到历史记录。

这是所有后续设计的地基。

早期代码是耦合的：

if new_digest!按理说，= old_digest:
send_email
store.save

至于三个问题。

1. 告警策略每变一次就要改一次主要检测逻辑。
2. send_email 抛异常会导致 store.save 不执行——通知失败连带丢数据。
3. 想回溯“如果当时阈值调低点会怎样”。做不到，因为没被通知的变更压根没记录。按理说，

拆开之后的分层：

抓取层 → 检测层 → 事件存储 → 投递层
Fetcher Detector EventStore Dispatcher Email/Webhook

检测层的唯一职责是如实记录。永远不判断“要不要告警”。

def check -> ChangeEvent | None:
snapshot = fetcher.fetch
prev = store.latest
store.save_snapshot # 无条件落盘，第一件事
if prev is None or prev.digest == snapshot.digest:
return None
event = ChangeEvent(
id=uuid4.hex。monitor_id=monitor.id,prev_id=prev.id,curr_id=snapshot.id,ratio=change_ratio,detected_at=datetime.now,)
store.save_event # 事件也无条件落盘
return event

投递层单独消费事件。不过，这样“要不要发”的所有逻辑都集中在一个地方。改起来不会碰到数据链路，

收益：使用者后来问“能不能按新阈值重新算一遍上个月漏掉的”。因为事件全存了这是一条 SQL 的事，不用重新抓取。

二、去重：同一个变更不要发两次

事件指纹

使用者痛点：网络抖动或多实例并发导致重复告警，使用者邮箱被刷屏。

用内容指纹而不是事件 ID 做去重键：

import hashlib
def dedupe_key -> str:
raw = f"{event.monitor_id}:{event.prev_digest}:{event.curr_digest}"
return hashlib.sha256).hexdigest

关键是用 而不是 curr_digest. 前者能正确处理 “A → B → A” 这种来回横跳：从 A 变到 B 是一次事件。 从 B 变回 A 是另一次事件，两次都该通知。怎么说呢，如果只用 curr_digest。第二次会被当成重复吃掉，

去重窗口使用带 TTL 的集合：

class DedupeWindow:
def __init__:
self.ttl = ttl_seconds
self._seen: dict = {}
def seen_recently -> bool:
now = time.monotonic
# 顺手清理过期项。避免无限增长
self._seen = {k: v for k,v in self._seen.items if now - v 

生产环境换成 Redis 的 SADD key NX EX ttl,一条命令搞定，还天然支持多实例。按理说，

抖动内容的识别

User Pain Point：A/B 测试页面频繁切换导致大量无意义告警。

A→B→A→B 每次都是合法变更，去重逻辑挡不住。处理办法是看近期指纹的基数:

def is_flapping -> bool:
"""最近 window 次检查里如果内容在 <= distinct_max 个状态间循环，判为抖动"""
    recent = store.recent_digests
    if len = window // 2

If 判定为抖动。就自动降级：停止推送，并发送一条“页面内容不稳定，已暂停告警”的元通知让使用者自行调节选择器或忽略规则。

三、限流：给每个维度都加上限

User Pain Point：a/b 大规模改版时短时间内产生海量告警，邮件箱和下游程序瞬间崩溃。不过，

限流要分层，因为不同层解决不同问题：

@dataclass
class RateLimits:
per_monitor_hourly: int = 20 # 单个监控项每小时最多多少条
per_user_hourly: int = 200 # 单使用者每小时最多多少条
per_channel_minute: int = 30 # 单通道每分钟最多多少条
def allowed -> tuple:
if counter.incr_and_get> limits.per_monitor_hourly:
return False,"monitor_hourly"
if counter.incr_and_get> limits.per_user_hourly:
return False。"user_hourly"
if counter.incr_and_get> limits.per_channel_minute:
return False,"channel_minute"
return True,""

Bounty: 被限流的事件不要丢失. 标记 `suppressed_reason` 存起来在窗口结束时发一条汇总：

四、静默窗口与聚合

免打扰时段

User Pain Point: 深夜大量非紧急告警让运维睡眠质量下降。

from zoneinfo import ZoneInfo
from datetime import time,datetime
@dataclass
class QuietHours:
再看start，time # 如：22:00
再看end，time # 如：07:00
再看tz。str = "Asia/Shanghai"
def contains -> bool:
local = dt.astimezone).time
if self.start <= self.end:
            return self.start <= local = self.start or local  datetime:
"""返回当前时间所在免打扰窗口结束时刻"""
today_end = dt.replace(hour=self.end.hour,minute=self.end.minute,second=0,microsecond=0)
if self.start <= self.end or dt.time 

Coding bug 常出现在跨午夜判断上；上述实现已覆盖该情形，

If 当前处于免打扰且不是关键事件，则把事件放入延迟队列：

if policy.quiet_hours.contains and not event.is_critical:
queue.defer)
return

`is_critical` 根据使用者配置的关键词决定。大多数使用者都会开启此开关，以确保真正关键的问题仍能及时触达。

聚合的粒度

User Pain Point: 一次性收到上百条相似邮件导致信息噪声爆炸。

• A simple “N 条拼成一条” 常常不可读；粒度选错反而更难读，
• The practical grouping key is ``.

This yields summaries such as “竞品价格：5 条变更 / 政策公告：3 条变更”，而不是“一锅乱炖”。

五、投递：重试、幂等、死信

Webhook 的重试要区分错误类型

User Pain Point: 盲目无限重试导致下游服务崩溃甚至网络拥塞。

RETRYABLE_STATUS = {429,500,502,503,504}
async def deliver(event: ChangeEvent,target: Webhook,attempt: int = 1) -> bool:
try的观点是，resp = await client.post(
target.url,json=event.to_payload,headers={
"X-Signature": sign,"X-Event-Id": event.id,"X-Delivery-Attempt": str,},timeout=10。)
except (httpx.TimeoutException,httpx.ConnectError):
return await schedule_retry(event,target,attempt)
# 成功响应
if 200 <= resp.status_code <300:
       return True
   # 可恢复错误
   if resp.status_code in RETRYABLE_STATUS:
       delay = int) or None
       return await schedule_retry(event,
                                   target,
                                   attempt,
                                   override_delay=delay)
   # 客户端错误 => 配置问题，不再重试
await mark_dead(event,target,reason=f"http_{resp.status_code}")
return False

// 不要对不可恢复错误进行重试// . URL 写错或密钥失效，只会把错误放大数倍。直接进入死信队列，并主动通知使用者 “你的 webhook 配置有问题”。

幂等键是给接收方的礼物

• `X-Event-Id` 能让对方安全去重，即使我们自己重试多次对方也只处理一次。没有这个字段，对方只能自行猜 key，在我们重试时极易翻车。

签名要包含时间戳

---