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