回調(diào)地址配好、消息能收到只是第一步。真正決定能不能上生產(chǎn)的是收到之后怎么處理——事件怎么分發(fā)、重復怎么擋、處理失敗了怎么補。這篇講接收之后的處理模式。一、收完先分層接入層不碰業(yè)務(wù)回調(diào)進來只做三件事驗簽、去重、入隊然后立刻返回。業(yè)務(wù)處理全部放到隊列后面。這一層不碰業(yè)務(wù)是關(guān)鍵。一旦在回調(diào)里查庫、調(diào)模型幾秒內(nèi)返回不了平臺判定失敗重推用戶就收到重復回復。二、事件分發(fā)路由表加處理器注冊事件類型一多用大 if 會失控。改成路由表——類型到處理器的映射新類型只注冊一個處理器不動已有代碼事件類型清單見Eyun 開發(fā)文檔。處理器按類型分消息事件、成員變動、狀態(tài)回調(diào)各一個互不干擾。這樣加功能很輕也方便單獨測試某個處理器。三、冪等、異步、補償三件事冪等按事件 ID 做處理前先查是否處理過處理過直接跳過。異步就是把隊列消費放到后臺線程或獨立進程不阻塞接收。補償是處理失敗后重試但要有次數(shù)上限超限進死信并告警。別做無限重試——重試到天荒地老問題還是那個問題。處理模式對照關(guān)注點做法關(guān)鍵點常見問題接入層驗簽去重入隊不碰業(yè)務(wù)回調(diào)里處理業(yè)務(wù)分發(fā)路由表處理器新類型只加注冊大 if 失控冪等按事件 ID 去重處理前先查重復處理補償限次重試死信超限告警無限重試回調(diào)處理實現(xiàn)import queue import threading class EventDispatcher: def __init__(self): self.handlers {} # 事件類型 - 處理器 self.queue queue.Queue() self.done set() threading.Thread(targetself._loop, daemonTrue).start() def register(self, event_type, handler): self.handlers[event_type] handler # 新類型只加這一行 def on_callback(self, event): # 接入層只驗簽、去重、入隊 eid event[event_id] if eid in self.done: return {code: 0} self.queue.put(event) return {code: 0} # 立即 ACK def _loop(self): while True: event self.queue.get() handler self.handlers.get(event[type]) if handler is None: continue # 未注冊的類型先跳過 for attempt in range(3): # 補償限次重試 try: handler(event) self.done.add(event[event_id]) break except Exception as e: if attempt 2: self.to_dead_letter(event, e) # 超限進死信落地建議回調(diào)里只做驗簽、去重、入隊業(yè)務(wù)放隊列后面ACK 才能及時。事件分發(fā)用路由表新類型只加注冊不改舊代碼。冪等按事件 ID 做重復推送直接跳過。失敗重試設(shè)次數(shù)上限超限進死信并告警別無限重試。接入層和業(yè)務(wù)層分開部署接收受平臺時間限制業(yè)務(wù)不受。微信側(cè)的回調(diào)推送由 Eyun 這類個人微信API平臺 提供分發(fā)與補償邏輯在自建服務(wù)實現(xiàn)具體字段以Eyun 平臺為準。