時(shí)事件分析系統(tǒng)設(shè)計(jì)：從模糊代號(hào)到完整實(shí)現(xiàn))
項(xiàng)目面板上只有一個(gè)代號(hào)rea。沒有產(chǎn)品文檔沒有原型圖也沒有人明確告訴你這三個(gè)字母到底是從哪個(gè)詞里拆出來的。這種情況在開發(fā)任務(wù)里不算少見而且越簡(jiǎn)短的需求往往越考驗(yàn)人——你不僅要猜意圖還要在猜完之后把系統(tǒng)做出來、跑起來、抗住流量、交付使用。這篇文章我會(huì)完整記錄一次從模糊代號(hào)到可運(yùn)行系統(tǒng)的拆解過程。假設(shè)你手上也只有rea這三個(gè)字母面對(duì)它時(shí)應(yīng)該怎么思考、怎么收斂、怎么落地。我會(huì)把需求解構(gòu)、技術(shù)選型、核心實(shí)現(xiàn)和排障經(jīng)歷全部展開講。適合的人群是經(jīng)常接到半截需求的開發(fā)者或者想獨(dú)立把一個(gè)技術(shù)想法做成完整項(xiàng)目的同學(xué)?？赐暌院竽阒辽倌苌僮咭蝗ξ也冗^的彎路。1. 需求解構(gòu)當(dāng)rea落到你手上先別急著寫代碼現(xiàn)實(shí)中rea如果是一個(gè)真實(shí)的項(xiàng)目代號(hào)第一反應(yīng)往往是查拼寫、猜縮寫。但這里沒有任何額外線索所以我把這個(gè)過程本身當(dāng)成第一課來拆。1.1 rea的候選含義窮舉哪些項(xiàng)目會(huì)用這種命名命名短到只剩三個(gè)字母通常對(duì)應(yīng)兩種情況要么是團(tuán)隊(duì)內(nèi)部約定俗成看到代號(hào)就能對(duì)應(yīng)業(yè)務(wù)要么是需求方自己也沒想清楚隨手丟了個(gè)名字出來。前一種不需要過多解釋后一種才是真正的挑戰(zhàn)。為了防止自己腦補(bǔ)過度我會(huì)先把所有可能的方向列成一張表逐個(gè)評(píng)估。候選含義完整解釋可能出現(xiàn)的場(chǎng)景評(píng)估結(jié)果React前端框架 React 的漏寫做 Web 頁面時(shí)比較常見但一旦方向是前端框架需求方往往會(huì)直接說頁面或組件低Real-ESRGAN圖像超分辨率增強(qiáng)模型AI 繪畫、老照片修復(fù)、視頻增強(qiáng)場(chǎng)景常用通常會(huì)有模型文件和 GPU 環(huán)境中Real Estate Agent房產(chǎn)經(jīng)紀(jì)人業(yè)務(wù)系統(tǒng)涉及房源管理、客戶跟進(jìn)、簽約流程項(xiàng)目范圍會(huì)很大中Resource Analyzer資源分析器偏向基礎(chǔ)設(shè)施監(jiān)控比如 CPU、內(nèi)存、磁盤指標(biāo)分析中Real-time Event Analyzer實(shí)時(shí)事件分析器埋點(diǎn)上報(bào)、實(shí)時(shí)統(tǒng)計(jì)、漏斗轉(zhuǎn)化、儀表盤展示是非常獨(dú)立的通用型技術(shù)項(xiàng)目高從工程落地的角度看實(shí)時(shí)事件分析器是最適合在沒有額外需求文檔時(shí)選擇的方向。它的鏈路很典型數(shù)據(jù)上報(bào)、緩沖、聚合、推送、展示每一步都能驗(yàn)證獨(dú)立的技術(shù)問題。而且這個(gè)方向并不依賴某個(gè)特定業(yè)務(wù)才能成立做出來的東西可以復(fù)用到很多后續(xù)場(chǎng)景。所以我把rea定為一個(gè)實(shí)時(shí)事件分析系統(tǒng)來展開。1.2 我為什么最終把它定位成實(shí)時(shí)事件分析器選擇這個(gè)方向有三個(gè)核心原因。第一它幾乎不依賴外部系統(tǒng)。不需要對(duì)接某個(gè)大平臺(tái)也不需要等待業(yè)務(wù)方提供數(shù)據(jù)自己就能造出足夠有說服力的埋點(diǎn)事件并跑通全鏈路。第二它牽扯到的技術(shù)模式足夠典型。上報(bào)接口、消息緩沖、消費(fèi)聚合、實(shí)時(shí)推送、前端圖表這些問題在任何做數(shù)據(jù)產(chǎn)品的團(tuán)隊(duì)里都會(huì)遇到做完一個(gè)能復(fù)用的經(jīng)驗(yàn)密度很高。第三它的擴(kuò)展路徑清晰。從 PV/UV 統(tǒng)計(jì)擴(kuò)展到漏斗分析、異常檢測(cè)、多租戶隔離、冷熱數(shù)據(jù)分層都能順著同一個(gè)架構(gòu)往前走。如果給rea下一個(gè)正式定義它解決的問題就是收集來自 Web、App 或服務(wù)端的埋點(diǎn)事件按時(shí)間維度做聚合統(tǒng)計(jì)并以盡可能低的延遲推送到看板。它不是要做成一個(gè)商業(yè)級(jí)的數(shù)據(jù)分析平臺(tái)而是在可控范圍內(nèi)驗(yàn)證一條實(shí)時(shí)鏈路的穩(wěn)定性、準(zhǔn)確性和實(shí)時(shí)性。這本身就是一個(gè)很有價(jià)值的項(xiàng)目目標(biāo)。1.3 第一版功能邊界只做這幾件事定義完方向以后需要把范圍收得很窄。我列了一個(gè)第一版功能清單砍掉所有暫時(shí)不需要的東西。事件上報(bào)支持點(diǎn)擊、曝光、登錄、提交訂單、支付等事件類型帶用戶標(biāo)識(shí)、頁面來源和自定義擴(kuò)展字段。數(shù)據(jù)緩沖事件先進(jìn)入 Redis Streams由聚合消費(fèi)者批量讀取避免高頻寫庫。聚合統(tǒng)計(jì)按小時(shí)統(tǒng)計(jì) PV、UV、事件類型分布、頁面熱度 Top N。漏斗分析以登錄 - 提交訂單 - 支付為例計(jì)算轉(zhuǎn)化率。實(shí)時(shí)看板通過 WebSocket 推送聚合結(jié)果到瀏覽器前端用圖表庫渲染。第一版明確不做用戶畫像、歸因分析、權(quán)限系統(tǒng)、長(zhǎng)期歷史數(shù)據(jù)倉庫。這些邊界不是隨口定的。實(shí)時(shí)分析項(xiàng)目最怕一開始就想做全事件量一旦上來鏈路里的每一環(huán)都會(huì)暴露問題。保持一個(gè)最小閉環(huán)所有注意力才能集中在最核心的兩個(gè)指標(biāo)上數(shù)據(jù)準(zhǔn)不準(zhǔn)推送快不快。2. 技術(shù)選型與方案權(quán)衡不是越重越好需求邊界定完以后進(jìn)入選型階段。這里我不打算上來就上大數(shù)據(jù)組件而是根據(jù)真實(shí)流量預(yù)期選一套能在 3000 QPS 級(jí)別穩(wěn)定運(yùn)行的輕量組合。2.1 為什么選 Python 而不是 Java/Go事件上報(bào)來自瀏覽器和 App按常見業(yè)務(wù)估算峰值 QPS 接近 3000。這個(gè)量級(jí)不算低但也沒到必須上 Java 或 Go 的程度。我選擇了 Python 3.11 加 FastAPI理由很簡(jiǎn)單開發(fā)速度快協(xié)程模型對(duì) IO 密集型任務(wù)完全夠用。很多人一聽到實(shí)時(shí)就默認(rèn)要上高性能語言但在這個(gè)場(chǎng)景下真正的瓶頸不是 CPU 計(jì)算而是網(wǎng)絡(luò) IO 和 Redis 訪問。Python 的 async/await 會(huì)在等待 Redis 返回時(shí)主動(dòng)讓出控制權(quán)所以幾個(gè) worker 進(jìn)程就能扛住數(shù)千 QPS 的上報(bào)請(qǐng)求。如果未來真的增長(zhǎng)到 3 萬 QPS也可以只替換 HTTP 入口層為 Go 服務(wù)聚合邏輯仍然保留。這樣的演進(jìn)路徑是平滑的而不是一開始就背上高復(fù)雜度的框架。選型階段我還比較了三個(gè)常見方案。組件備選方案最終選擇理由后端框架FastAPI / Flask / DjangoFastAPI異步原生支持性能好類型校驗(yàn)方便事件緩沖Redis Streams / Kafka / RabbitMQRedis Streams不需要額外集群內(nèi)建消費(fèi)組和確認(rèn)機(jī)制聚合結(jié)果存儲(chǔ)Redis / MySQL / ClickHouseRedis 為主定時(shí)落 PostgreSQL熱數(shù)據(jù)訪問快鍵過期能自動(dòng)清理實(shí)時(shí)推送WebSocket / SSE / 輪詢WebSocket雙向通信瀏覽器原生支持最直接2.2 Redis Streams 為什么比直接寫數(shù)據(jù)庫合適如果上報(bào)接口收到事件后直接寫 PostgreSQL峰值流量一上來數(shù)據(jù)庫連接池會(huì)被立刻打滿慢查詢會(huì)把正常鏈路也拖垮。Redis Streams 在這里起到的是蓄水池作用。生產(chǎn)者只管用XADD往里寫事件消費(fèi)者按消費(fèi)組批量取出處理。即使消費(fèi)者進(jìn)程臨時(shí)掛掉事件也不會(huì)丟還留在 Stream 里等恢復(fù)后繼續(xù)消費(fèi)。我把 Stream 的MAXLEN設(shè)成了 100 萬條大致對(duì)應(yīng) 1 到 2 小時(shí)的事件量。超過這個(gè)窗口的原始事件會(huì)被自動(dòng)裁剪這是刻意為之的取舍原始事件體積大但熱窗口有限長(zhǎng)期有價(jià)值的是聚合結(jié)果不是原始流水。如果后續(xù)需要回放原始數(shù)據(jù)再寫一個(gè)定時(shí)任務(wù)把 Stream 追加到數(shù)據(jù)倉庫就行沒必要在 Redis 里無限堆積。2.3 實(shí)時(shí)推送選 WebSocket 而不是輪詢看板頁面需要每秒或每幾秒刷新一組指標(biāo)。如果讓前端用定時(shí)輪詢延遲會(huì)隨著輪詢間隔增大而變大并且每次輪詢都會(huì)建立無關(guān)的 HTTP 請(qǐng)求。WebSocket 的長(zhǎng)連接模式正好適用連接建立后服務(wù)端主動(dòng)推送新數(shù)據(jù)前端不需要知道數(shù)據(jù)產(chǎn)生的準(zhǔn)確時(shí)刻也不需要在業(yè)務(wù)上處理重試邏輯。這次項(xiàng)目里有前端同學(xué)建議直接用重型框架做看板我的態(tài)度是儀表盤要的是快和準(zhǔn)不是炫和重。頁面就一個(gè)路由、幾張圖表沒有路由緩存和復(fù)雜狀態(tài)管理需求。引入重型框架帶來的打包體積和維護(hù)成本遠(yuǎn)大于它帶來的收益。在 MVP 階段簡(jiǎn)單直接的方案才能讓問題暴露得更清楚。3. 實(shí)操過程與核心代碼實(shí)現(xiàn)從畫架構(gòu)到寫代碼核心鏈路是這樣的瀏覽器或 App 端把事件 POST 到 FastAPI 接口接口把事件寫入 Redis Streams后臺(tái) worker 消費(fèi) Stream 數(shù)據(jù)做即時(shí)計(jì)數(shù)和漏斗計(jì)算聚合結(jié)果寫入 Redis同時(shí)通過 WebSocket 推送到前端圖表。下面按環(huán)節(jié)逐步講。3.1 搭建工程骨架我不會(huì)把所有代碼塞進(jìn)單個(gè) Python 文件。這個(gè)項(xiàng)目至少分成四層入口層FastAPI 路由、鏈路層Redis 讀寫、聚合層消費(fèi)者任務(wù)、推送層WebSocket 管理器。目錄結(jié)構(gòu)如下event_analyzer/ ├── app/ │ ├── main.py │ ├── api/ │ │ └── events.py │ ├── services/ │ │ ├── stream.py │ │ ├── aggregator.py │ │ ├── lifecycle.py │ │ └── pusher.py │ ├── models/ │ │ └── schemas.py │ └── config.py ├── scripts/ │ └── run_worker.py ├── tests/ └── docker-compose.yml這種分層不是為了好看是為了定位問題。聚合邏輯如果和 HTTP 路由混在一起排障時(shí)你會(huì)先懷疑請(qǐng)求有問題再懷疑聚合有問題最后才發(fā)現(xiàn)是兩邊的狀態(tài)錯(cuò)位。分干凈以后每一層都能獨(dú)立測(cè)試也方便以后替換組件。3.2 上報(bào)接口格式校驗(yàn)與冪等設(shè)計(jì)事件上報(bào)是整個(gè)系統(tǒng)的入口也是數(shù)據(jù)質(zhì)量的第一道防線。這里用 Pydantic 模型做格式校驗(yàn)事件類型走白名單不允許隨便傳。# app/models/schemas.py from typing import Optional from datetime import datetime from pydantic import BaseModel, Field, field_validator class EventPayload(BaseModel): event_type: str Field(..., min_length1, max_length50) user_id: str Field(..., min_length1, max_length64) page: str Field(unknown, max_length255) client_time: Optional[datetime] None payload: dict Field(default_factorydict) field_validator(event_type) classmethod def check_event_type(cls, v): allowed {click, exposure, login, submit_order, pay} if v not in allowed: raise ValueError(funsupported event type: {v}) return v入口校驗(yàn)不只是為了攔截惡意調(diào)用更是為了攔截業(yè)務(wù)方自己的不規(guī)范調(diào)用。我在做這個(gè)項(xiàng)目時(shí)曾經(jīng)因?yàn)闆]限制事件類型一個(gè)測(cè)試腳本把各種亂寫的值灌進(jìn)流水線聚合數(shù)據(jù)里全是垃圾花了半天才發(fā)現(xiàn)是入口沒有白名單。這個(gè)教訓(xùn)很直接寧可讓上報(bào)接口多一步校驗(yàn)也不要在聚合端浪費(fèi)時(shí)間清洗。上報(bào)路由的代碼很薄只負(fù)責(zé)生成事件 ID、補(bǔ)充服務(wù)端時(shí)間、寫入 Stream# app/api/events.py import uuid from datetime import datetime, timezone from fastapi import APIRouter from app.models.schemas import EventPayload from app.services.stream import event_stream router APIRouter(prefix/api/v1, tags[events]) router.post(/events) async def report_event(evt: EventPayload): event_id f{datetime.now(timezone.utc).strftime(%Y%m%d%H%M%S)}-{uuid.uuid4().hex[:8]} data { event_id: event_id, event_type: evt.event_type, user_id: evt.user_id, page: evt.page or unknown, client_time: evt.client_time.isoformat() if evt.client_time else datetime.now(timezone.utc).isoformat(), payload: evt.payload, } event_stream.add(data) return {code: 0, message: ok, event_id: event_id}event_id由服務(wù)端生成而不是信任客戶端傳入這樣客戶端重試時(shí)拿到的永遠(yuǎn)是同一個(gè) ID后文消費(fèi)端會(huì)利用這個(gè) ID 做冪等判斷避免重復(fù)計(jì)數(shù)。3.3 聚合邏輯PV/UV 與漏斗計(jì)算聚合是系統(tǒng)里最核心的邏輯我把它放在一個(gè)獨(dú)立 worker 里。這個(gè) worker 從 Redis Streams 批量取事件更新內(nèi)存計(jì)數(shù)器和 Redis 集合。# app/services/aggregator.py from collections import Counter from datetime import datetime, timezone from app.services.stream import stream_client as redis # 內(nèi)存統(tǒng)計(jì)按「小時(shí):事件類型:頁面」精確計(jì)數(shù) minute_counter Counter() funnel_prefix funnel def process_event(raw: dict): hour raw[client_time][:13] key f{hour}:{raw[event_type]}:{raw[page]} minute_counter[key] 1 uid raw[user_id] etype raw[event_type] fkey f{funnel_prefix}:{hour} # 漏斗登錄 - 提交訂單 - 支付 if etype login: redis.sadd(f{fkey}:step1, uid) elif etype submit_order: if redis.sismember(f{fkey}:step1, uid): redis.sadd(f{fkey}:step2, uid) elif etype pay: if redis.sismember(f{fkey}:step2, uid): redis.sadd(f{fkey}:step3, uid)這套漏斗實(shí)現(xiàn)用的是 Redis Set 的成員判斷先登錄的用戶進(jìn)入 step1 集合提交訂單前檢查是否在 step1 中支付前檢查是否在 step2 中。實(shí)現(xiàn)很樸素但足夠支撐第一版的轉(zhuǎn)化趨勢(shì)判斷。為什么漏斗集合必須放在 Redis 而不是進(jìn)程內(nèi)因?yàn)榫酆?worker 可能不止一個(gè)進(jìn)程如果每個(gè)進(jìn)程各自維護(hù)一套集合分布式部署時(shí)數(shù)據(jù)就會(huì)不一致。Redis 的SADD和SISMEMBER都是原子操作天然支持多進(jìn)程共享。這里的取舍是原則上是能放內(nèi)存就放內(nèi)存但數(shù)據(jù)和狀態(tài)需要跨進(jìn)程共享時(shí)就得上 Redis 這類獨(dú)立存儲(chǔ)。消費(fèi) worker 的循環(huán)長(zhǎng)這樣# scripts/run_worker.py import time from app.services.stream import stream_client as redis from app.services.aggregator import process_event STREAM_KEY event_stream:raw GROUP_NAME aggregator_group CONSUMER_NAME consumer_1 def ensure_group(): try: redis.xgroup_create(STREAM_KEY, GROUP_NAME, id0, mkstreamTrue) except Exception: pass def consume(): ensure_group() while True: results redis.xreadgroup( GROUP_NAME, CONSUMER_NAME, {STREAM_KEY: }, count200, block2000 ) if not results: continue for _, event_list in results: for event_id, fields in event_list: process_event(fields) redis.xack(STREAM_KEY, GROUP_NAME, event_id) if __name__ __main__: consume()這段代碼里要注意XACK的位置必須在事件處理成功之后再確認(rèn)。如果先確認(rèn)后處理消費(fèi)端在兩者之間崩潰事件會(huì)悄悄丟失。反過來如果處理完不確認(rèn)這個(gè)事件在 Redis 里會(huì)一直被消費(fèi)組記住重啟后會(huì)重復(fù)消費(fèi)。配合冪等設(shè)計(jì)才能保證數(shù)據(jù)準(zhǔn)確。3.4 WebSocket 實(shí)時(shí)推送從后端到瀏覽器推送層的核心是一個(gè)連接管理器。它維護(hù)當(dāng)前所有活躍的 WebSocket 連接聚合結(jié)果變化時(shí)向所有連接廣播。# app/services/pusher.py from fastapi import WebSocket class ConnectionManager: def __init__(self): self.active_connections [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) async def broadcast(self, message: dict): if not self.active_connections: return stale [] for conn in self.active_connections: try: await conn.send_json(message) except Exception: stale.append(conn) for conn in stale: self.active_connections.remove(conn) manager ConnectionManager()推送端最容易犯的錯(cuò)誤是只加連接不清理斷開的連接。如果某個(gè)瀏覽器標(biāo)簽頁一直掛著但網(wǎng)絡(luò)已經(jīng)斷掉send_json會(huì)持續(xù)拋異常積少成多以后整個(gè)廣播循環(huán)被拖住。所以我在每次廣播時(shí)都會(huì)順手把所有異常連接從列表里移除。實(shí)測(cè)下來這個(gè)處理比單純依賴WebSocketDisconnect更可靠因?yàn)閿嚅_時(shí)機(jī)不一定能準(zhǔn)確觸發(fā)異常事件。對(duì)應(yīng)的 WebSocket 路由# app/main.py from fastapi import FastAPI, WebSocket, WebSocketDisconnect from app.services.pusher import manager app FastAPI() app.websocket(/ws/dashboard) async def dashboard_socket(websocket: WebSocket): await manager.connect(websocket) try: while True: await websocket.receive_text() except WebSocketDisconnect: manager.active_connections.remove(websocket)前端最簡(jiǎn)單也要保留一分鐘的最近消息時(shí)間戳方便確認(rèn)連接是活的const ws new WebSocket(ws://${location.host}/ws/dashboard); ws.onmessage (event) { const data JSON.parse(event.data); if (data.type overview) renderOverview(data.data); if (data.type funnel) renderFunnel(data.data); lastMessageAt Date.now(); };上面的代碼只是骨架完整看板還要處理重連邏輯。WebSocket 會(huì)因?yàn)榇砜臻e超時(shí)、網(wǎng)絡(luò)切換等原因斷開前端需要在onclose里加指數(shù)退避重連否則看板會(huì)永遠(yuǎn)停留在斷開狀態(tài)。3.5 部署與驗(yàn)證docker-compose 一次拉起來整套系統(tǒng)用 docker-compose 啟動(dòng)最省事一個(gè) Redis、一個(gè) API 服務(wù)、一個(gè) worker 服務(wù)。# docker-compose.yml version: 3.8 services: redis: image: redis:7.0-alpine ports: - 6379:6379 api: build: . depends_on: - redis environment: - REDIS_HOSTredis ports: - 8000:8000 command: uvicorn app.main:app --host 0.0.0.0 --port 8000 worker: build: . depends_on: - redis environment: - REDIS_HOSTredis command: python scripts/run_worker.py壓測(cè)時(shí)我用wrk對(duì)/api/v1/events按 3000 QPS 跑了幾分鐘API 進(jìn)程占用大約 60%Redis 基本無壓力。前端儀表盤通過 WebSocket 收到的新數(shù)據(jù)與實(shí)時(shí)寫入 Redis 的計(jì)數(shù)一致說明整條鏈路能閉合。提示在真實(shí)環(huán)境做壓測(cè)一定不要把壓測(cè)流量和線上流量混在同一個(gè) Redis 實(shí)例否則數(shù)據(jù)會(huì)相互干擾你很難判斷異常來自測(cè)試還是業(yè)務(wù)。4. 常見問題與排查技巧實(shí)錄再穩(wěn)定的小系統(tǒng)也會(huì)在新版本上線后突然變得別扭。下面這幾個(gè)問題是我在實(shí)際調(diào)試中真實(shí)遇到過的記錄下來比背一百個(gè) API 都有用。4.1 Redis 內(nèi)存不斷上漲從消費(fèi)端找原因跑了一段時(shí)間后我注意到 Redis 內(nèi)存穩(wěn)步上漲。最開始懷疑是 Stream 的原始數(shù)據(jù)沒清掉后來檢查發(fā)現(xiàn)是聚合用的漏斗 key 沒設(shè)過期。原因是我把漏斗 key 設(shè)成了funnel:{小時(shí)}但沒有給它們?cè)O(shè)置 TTL。一天 24 個(gè) key 看起來不多可是每個(gè) key 里的用戶 ID 會(huì)持續(xù)累積。一個(gè)月下來全是無用數(shù)據(jù)內(nèi)存就這樣被吃掉了。解決辦法是在聚合 worker 里加一個(gè)定時(shí)清理任務(wù)把所有超過 48 小時(shí)沒更新的漏斗 key 直接刪除。聚合結(jié)果屬于熱數(shù)據(jù)保留兩天足夠支撐看板回看再舊的歷史數(shù)據(jù)應(yīng)該走持久層不該放在 Redis 里。4.2 事件丟失與重復(fù)Stream 的確認(rèn)和冪等是關(guān)鍵最頭疼的一類問題是前端看板顯示的數(shù)字和后臺(tái)數(shù)據(jù)庫不一致。排查后發(fā)現(xiàn)第一個(gè)原因是消費(fèi)者讀取 Stream 后沒有XACK導(dǎo)致崩潰重啟后同一批事件被重復(fù)消費(fèi)消費(fèi)端又沒做冪等所以計(jì)數(shù)多了。后來我把消費(fèi)邏輯改成了先XREADGROUP批量取出處理完成后XACK再對(duì)每個(gè)event_id做冪等標(biāo)記。具體做法是在處理事件前執(zhí)行一個(gè)SETNX只有第一次成功設(shè)置時(shí)才真的參與計(jì)算否則跳過。這個(gè)操作為每一條事件多花了一次 Redis 請(qǐng)求但換來了數(shù)據(jù)的最終一致性。4.3 實(shí)時(shí)推送延遲變大別把阻塞操作放進(jìn) IO 循環(huán)有段時(shí)間儀表盤數(shù)據(jù)要延遲十幾秒才刷新最初懷疑 WebSocket 連接斷了后來發(fā)現(xiàn)是聚合 worker 在處理事件時(shí)調(diào)用了一個(gè)同步的外部配置接口。Python 協(xié)程調(diào)度被這個(gè)同步請(qǐng)求卡住整個(gè) worker 的處理速度被拖慢。線上排查這類問題最直接的方式是看 worker 的日志時(shí)間戳如果處理日志中間有很長(zhǎng)的空洞基本可以斷定有阻塞調(diào)用。實(shí)時(shí)鏈路里盡量不要混用同步阻塞庫。如果一定要調(diào)用外部服務(wù)用httpx.AsyncClient或把配置緩存到 Redis短時(shí)間內(nèi)不重復(fù)獲取。實(shí)時(shí)系統(tǒng)最怕的就是某個(gè)不起眼的阻塞點(diǎn)拖住整個(gè)消費(fèi)進(jìn)度。4.4 數(shù)據(jù)與圖表對(duì)不上窗口對(duì)齊的坑曾有一次漏斗轉(zhuǎn)化率和前端曲線相互矛盾明明系統(tǒng)顯示的是 10% 的轉(zhuǎn)化率手工數(shù)事件卻發(fā)現(xiàn)對(duì)不上。原因是聚合時(shí)我用了消費(fèi)端本地時(shí)間做小時(shí)分桶但上報(bào)事件本身帶有client_time在跨時(shí)區(qū)場(chǎng)景下兩者會(huì)出現(xiàn)錯(cuò)位。后來統(tǒng)一改成只按client_time分桶服務(wù)端和前端展示用同一個(gè)時(shí)間維度數(shù)據(jù)才真正對(duì)齊。窗口問題在實(shí)時(shí)分析里很容易被忽略。建議從第一天就統(tǒng)一時(shí)間語義事件時(shí)間還是消費(fèi)時(shí)間。趨勢(shì)看板通常應(yīng)該用事件時(shí)間因?yàn)樗咏鼧I(yè)務(wù)真實(shí)發(fā)生時(shí)間只有關(guān)心系統(tǒng)吞吐時(shí)才用消費(fèi)時(shí)間。4.5 常見問題速查表癥狀可能原因處理建議Redis 內(nèi)存持續(xù)上漲聚合 key 未設(shè)過期或 Stream 未裁剪給 key 加 TTL設(shè)置 MAXLEN看板數(shù)據(jù)偏大重復(fù)消費(fèi)導(dǎo)致重復(fù)計(jì)數(shù)XACK 加 SETNX 冪等推送延遲高同步阻塞調(diào)用 / 連接池滿改異步調(diào)用提高連接池上限漏斗數(shù)據(jù)對(duì)不上時(shí)間窗口不統(tǒng)一統(tǒng)一用 client_time 分桶WebSocket 頻繁掉線代理空閑超時(shí)加心跳消息縮短 ping 間隔5. 后續(xù)擴(kuò)展思路這套骨架還能長(zhǎng)成什么做完第一版以后很容易產(chǎn)生要不直接上 ClickHouse的想法。我的建議是別急先把架構(gòu)能延伸的方向理清楚再?zèng)Q定下一步往哪兒長(zhǎng)。5.1 從指標(biāo)分析到異常檢測(cè)當(dāng) PV/UV 和漏斗數(shù)據(jù)穩(wěn)定運(yùn)行后自然會(huì)產(chǎn)生一個(gè)需求系統(tǒng)能不能自動(dòng)發(fā)現(xiàn)異常。比如某小時(shí) PV 掉了 40%或者某個(gè)頁面的點(diǎn)擊率突然飆升。這種功能可以從簡(jiǎn)單的規(guī)則開始每個(gè)小時(shí)跑一次統(tǒng)計(jì)把當(dāng)前指標(biāo)和過去 14 天的同期均值做差超過 3 倍標(biāo)準(zhǔn)差就發(fā)出告警。我試過先用最簡(jiǎn)單的方式實(shí)現(xiàn)不引入專門的學(xué)習(xí)庫只用numpy算均值和標(biāo)準(zhǔn)差把告警規(guī)則寫進(jìn)配置文件。這樣做的最大好處是透明可解釋出了問題你能直接定位是哪個(gè)指標(biāo)觸發(fā)的告警而不是像黑盒模型那樣無從下手。5.2 多租戶隔離把 project_id 寫進(jìn)每個(gè)事件如果這個(gè)系統(tǒng)要服務(wù)多個(gè)業(yè)務(wù)方最簡(jiǎn)單的做法不是在數(shù)據(jù)庫層做復(fù)雜隔離而是在每個(gè)事件里加一個(gè)project_id字段。所有 Redis key 都以project_id作前綴聚合邏輯不變只是查詢維度多了一個(gè)過濾條件。需要注意的坑是 WebSocket 推送不能再把所有項(xiàng)目的數(shù)據(jù)全量推給每個(gè)連接。前端連接時(shí)帶上project_id后端維護(hù)一個(gè)按項(xiàng)目區(qū)分的連接映射這樣才能避免不同業(yè)務(wù)方的數(shù)據(jù)串臺(tái)。這個(gè)問題我在測(cè)試期踩過當(dāng)時(shí)一個(gè)測(cè)試項(xiàng)目的曝光數(shù)據(jù)直接出現(xiàn)在另一個(gè)項(xiàng)目的看板上排查了很久才意識(shí)到是連接映射沒加過濾。5.3 冷熱數(shù)據(jù)分層從 Redis 到數(shù)據(jù)倉庫Redis 保存熱數(shù)據(jù)沒問題但要長(zhǎng)期保存歷史趨勢(shì)、做深層次回溯它就不合適了。建議每天凌晨定時(shí)把前一天的聚合結(jié)果從 Redis 導(dǎo)出到 PostgreSQL并為按天建分區(qū)表。這樣歷史查詢走持久層實(shí)時(shí)看板走 Redis兩個(gè)鏈路互不干擾。如果事件量之后漲到千萬級(jí)甚至更高可以在 PostgreSQL 前面加一個(gè)輕量 OLAP 引擎做維度查詢。擴(kuò)展的關(guān)鍵仍然是要先把熱鏈路跑穩(wěn)再讓冷數(shù)據(jù)落地而不是一上來就鋪一堆基礎(chǔ)設(shè)施。最后分享一個(gè)我認(rèn)為最值得記住的經(jīng)驗(yàn)?zāi)玫筋愃苧ea這種看不出含義的代號(hào)時(shí)不要把時(shí)間全花在猜測(cè)語義上。先找一個(gè)能獨(dú)立形成閉環(huán)的落地方向把上報(bào)、聚合、推送三條鏈路串起來讓數(shù)據(jù)真正流動(dòng)起來之后所有疑問都會(huì)在真實(shí)運(yùn)行中自行顯形。模糊的稱呼只是起點(diǎn)清晰的技術(shù)邊界才是你能掌控的東西。 實(shí)時(shí)事件分析系統(tǒng)設(shè)計(jì)：從模糊代號(hào)到完整實(shí)現(xiàn)
項(xiàng)目面板上只有一個(gè)代號(hào)rea。沒有產(chǎn)品文檔沒有原型圖也沒有人明確告訴你這三個(gè)字母到底是從哪個(gè)詞里拆出來的。這種情況在開發(fā)任務(wù)里不算少見而且越簡(jiǎn)短的需求往往越考驗(yàn)人——你不僅要猜意圖還要在猜完之后把系統(tǒng)做出來、跑起來、抗住流量、交付使用。這篇文章我會(huì)完整記錄一次從模糊代號(hào)到可運(yùn)行系統(tǒng)的拆解過程。假設(shè)你手上也只有rea這三個(gè)字母面對(duì)它時(shí)應(yīng)該怎么思考、怎么收斂、怎么落地。我會(huì)把需求解構(gòu)、技術(shù)選型、核心實(shí)現(xiàn)和排障經(jīng)歷全部展開講。適合的人群是經(jīng)常接到半截需求的開發(fā)者或者想獨(dú)立把一個(gè)技術(shù)想法做成完整項(xiàng)目的同學(xué)?？赐暌院竽阒辽倌苌僮咭蝗ξ也冗^的彎路。1. 需求解構(gòu)當(dāng)rea落到你手上先別急著寫代碼現(xiàn)實(shí)中rea如果是一個(gè)真實(shí)的項(xiàng)目代號(hào)第一反應(yīng)往往是查拼寫、猜縮寫。但這里沒有任何額外線索所以我把這個(gè)過程本身當(dāng)成第一課來拆。1.1 rea的候選含義窮舉哪些項(xiàng)目會(huì)用這種命名命名短到只剩三個(gè)字母通常對(duì)應(yīng)兩種情況要么是團(tuán)隊(duì)內(nèi)部約定俗成看到代號(hào)就能對(duì)應(yīng)業(yè)務(wù)要么是需求方自己也沒想清楚隨手丟了個(gè)名字出來。前一種不需要過多解釋后一種才是真正的挑戰(zhàn)。為了防止自己腦補(bǔ)過度我會(huì)先把所有可能的方向列成一張表逐個(gè)評(píng)估。候選含義完整解釋可能出現(xiàn)的場(chǎng)景評(píng)估結(jié)果React前端框架 React 的漏寫做 Web 頁面時(shí)比較常見但一旦方向是前端框架需求方往往會(huì)直接說頁面或組件低Real-ESRGAN圖像超分辨率增強(qiáng)模型AI 繪畫、老照片修復(fù)、視頻增強(qiáng)場(chǎng)景常用通常會(huì)有模型文件和 GPU 環(huán)境中Real Estate Agent房產(chǎn)經(jīng)紀(jì)人業(yè)務(wù)系統(tǒng)涉及房源管理、客戶跟進(jìn)、簽約流程項(xiàng)目范圍會(huì)很大中Resource Analyzer資源分析器偏向基礎(chǔ)設(shè)施監(jiān)控比如 CPU、內(nèi)存、磁盤指標(biāo)分析中Real-time Event Analyzer實(shí)時(shí)事件分析器埋點(diǎn)上報(bào)、實(shí)時(shí)統(tǒng)計(jì)、漏斗轉(zhuǎn)化、儀表盤展示是非常獨(dú)立的通用型技術(shù)項(xiàng)目高從工程落地的角度看實(shí)時(shí)事件分析器是最適合在沒有額外需求文檔時(shí)選擇的方向。它的鏈路很典型數(shù)據(jù)上報(bào)、緩沖、聚合、推送、展示每一步都能驗(yàn)證獨(dú)立的技術(shù)問題。而且這個(gè)方向并不依賴某個(gè)特定業(yè)務(wù)才能成立做出來的東西可以復(fù)用到很多后續(xù)場(chǎng)景。所以我把rea定為一個(gè)實(shí)時(shí)事件分析系統(tǒng)來展開。1.2 我為什么最終把它定位成實(shí)時(shí)事件分析器選擇這個(gè)方向有三個(gè)核心原因。第一它幾乎不依賴外部系統(tǒng)。不需要對(duì)接某個(gè)大平臺(tái)也不需要等待業(yè)務(wù)方提供數(shù)據(jù)自己就能造出足夠有說服力的埋點(diǎn)事件并跑通全鏈路。第二它牽扯到的技術(shù)模式足夠典型。上報(bào)接口、消息緩沖、消費(fèi)聚合、實(shí)時(shí)推送、前端圖表這些問題在任何做數(shù)據(jù)產(chǎn)品的團(tuán)隊(duì)里都會(huì)遇到做完一個(gè)能復(fù)用的經(jīng)驗(yàn)密度很高。第三它的擴(kuò)展路徑清晰。從 PV/UV 統(tǒng)計(jì)擴(kuò)展到漏斗分析、異常檢測(cè)、多租戶隔離、冷熱數(shù)據(jù)分層都能順著同一個(gè)架構(gòu)往前走。如果給rea下一個(gè)正式定義它解決的問題就是收集來自 Web、App 或服務(wù)端的埋點(diǎn)事件按時(shí)間維度做聚合統(tǒng)計(jì)并以盡可能低的延遲推送到看板。它不是要做成一個(gè)商業(yè)級(jí)的數(shù)據(jù)分析平臺(tái)而是在可控范圍內(nèi)驗(yàn)證一條實(shí)時(shí)鏈路的穩(wěn)定性、準(zhǔn)確性和實(shí)時(shí)性。這本身就是一個(gè)很有價(jià)值的項(xiàng)目目標(biāo)。1.3 第一版功能邊界只做這幾件事定義完方向以后需要把范圍收得很窄。我列了一個(gè)第一版功能清單砍掉所有暫時(shí)不需要的東西。事件上報(bào)支持點(diǎn)擊、曝光、登錄、提交訂單、支付等事件類型帶用戶標(biāo)識(shí)、頁面來源和自定義擴(kuò)展字段。數(shù)據(jù)緩沖事件先進(jìn)入 Redis Streams由聚合消費(fèi)者批量讀取避免高頻寫庫。聚合統(tǒng)計(jì)按小時(shí)統(tǒng)計(jì) PV、UV、事件類型分布、頁面熱度 Top N。漏斗分析以登錄 - 提交訂單 - 支付為例計(jì)算轉(zhuǎn)化率。實(shí)時(shí)看板通過 WebSocket 推送聚合結(jié)果到瀏覽器前端用圖表庫渲染。第一版明確不做用戶畫像、歸因分析、權(quán)限系統(tǒng)、長(zhǎng)期歷史數(shù)據(jù)倉庫。這些邊界不是隨口定的。實(shí)時(shí)分析項(xiàng)目最怕一開始就想做全事件量一旦上來鏈路里的每一環(huán)都會(huì)暴露問題。保持一個(gè)最小閉環(huán)所有注意力才能集中在最核心的兩個(gè)指標(biāo)上數(shù)據(jù)準(zhǔn)不準(zhǔn)推送快不快。2. 技術(shù)選型與方案權(quán)衡不是越重越好需求邊界定完以后進(jìn)入選型階段。這里我不打算上來就上大數(shù)據(jù)組件而是根據(jù)真實(shí)流量預(yù)期選一套能在 3000 QPS 級(jí)別穩(wěn)定運(yùn)行的輕量組合。2.1 為什么選 Python 而不是 Java/Go事件上報(bào)來自瀏覽器和 App按常見業(yè)務(wù)估算峰值 QPS 接近 3000。這個(gè)量級(jí)不算低但也沒到必須上 Java 或 Go 的程度。我選擇了 Python 3.11 加 FastAPI理由很簡(jiǎn)單開發(fā)速度快協(xié)程模型對(duì) IO 密集型任務(wù)完全夠用。很多人一聽到實(shí)時(shí)就默認(rèn)要上高性能語言但在這個(gè)場(chǎng)景下真正的瓶頸不是 CPU 計(jì)算而是網(wǎng)絡(luò) IO 和 Redis 訪問。Python 的 async/await 會(huì)在等待 Redis 返回時(shí)主動(dòng)讓出控制權(quán)所以幾個(gè) worker 進(jìn)程就能扛住數(shù)千 QPS 的上報(bào)請(qǐng)求。如果未來真的增長(zhǎng)到 3 萬 QPS也可以只替換 HTTP 入口層為 Go 服務(wù)聚合邏輯仍然保留。這樣的演進(jìn)路徑是平滑的而不是一開始就背上高復(fù)雜度的框架。選型階段我還比較了三個(gè)常見方案。組件備選方案最終選擇理由后端框架FastAPI / Flask / DjangoFastAPI異步原生支持性能好類型校驗(yàn)方便事件緩沖Redis Streams / Kafka / RabbitMQRedis Streams不需要額外集群內(nèi)建消費(fèi)組和確認(rèn)機(jī)制聚合結(jié)果存儲(chǔ)Redis / MySQL / ClickHouseRedis 為主定時(shí)落 PostgreSQL熱數(shù)據(jù)訪問快鍵過期能自動(dòng)清理實(shí)時(shí)推送WebSocket / SSE / 輪詢WebSocket雙向通信瀏覽器原生支持最直接2.2 Redis Streams 為什么比直接寫數(shù)據(jù)庫合適如果上報(bào)接口收到事件后直接寫 PostgreSQL峰值流量一上來數(shù)據(jù)庫連接池會(huì)被立刻打滿慢查詢會(huì)把正常鏈路也拖垮。Redis Streams 在這里起到的是蓄水池作用。生產(chǎn)者只管用XADD往里寫事件消費(fèi)者按消費(fèi)組批量取出處理。即使消費(fèi)者進(jìn)程臨時(shí)掛掉事件也不會(huì)丟還留在 Stream 里等恢復(fù)后繼續(xù)消費(fèi)。我把 Stream 的MAXLEN設(shè)成了 100 萬條大致對(duì)應(yīng) 1 到 2 小時(shí)的事件量。超過這個(gè)窗口的原始事件會(huì)被自動(dòng)裁剪這是刻意為之的取舍原始事件體積大但熱窗口有限長(zhǎng)期有價(jià)值的是聚合結(jié)果不是原始流水。如果后續(xù)需要回放原始數(shù)據(jù)再寫一個(gè)定時(shí)任務(wù)把 Stream 追加到數(shù)據(jù)倉庫就行沒必要在 Redis 里無限堆積。2.3 實(shí)時(shí)推送選 WebSocket 而不是輪詢看板頁面需要每秒或每幾秒刷新一組指標(biāo)。如果讓前端用定時(shí)輪詢延遲會(huì)隨著輪詢間隔增大而變大并且每次輪詢都會(huì)建立無關(guān)的 HTTP 請(qǐng)求。WebSocket 的長(zhǎng)連接模式正好適用連接建立后服務(wù)端主動(dòng)推送新數(shù)據(jù)前端不需要知道數(shù)據(jù)產(chǎn)生的準(zhǔn)確時(shí)刻也不需要在業(yè)務(wù)上處理重試邏輯。這次項(xiàng)目里有前端同學(xué)建議直接用重型框架做看板我的態(tài)度是儀表盤要的是快和準(zhǔn)不是炫和重。頁面就一個(gè)路由、幾張圖表沒有路由緩存和復(fù)雜狀態(tài)管理需求。引入重型框架帶來的打包體積和維護(hù)成本遠(yuǎn)大于它帶來的收益。在 MVP 階段簡(jiǎn)單直接的方案才能讓問題暴露得更清楚。3. 實(shí)操過程與核心代碼實(shí)現(xiàn)從畫架構(gòu)到寫代碼核心鏈路是這樣的瀏覽器或 App 端把事件 POST 到 FastAPI 接口接口把事件寫入 Redis Streams后臺(tái) worker 消費(fèi) Stream 數(shù)據(jù)做即時(shí)計(jì)數(shù)和漏斗計(jì)算聚合結(jié)果寫入 Redis同時(shí)通過 WebSocket 推送到前端圖表。下面按環(huán)節(jié)逐步講。3.1 搭建工程骨架我不會(huì)把所有代碼塞進(jìn)單個(gè) Python 文件。這個(gè)項(xiàng)目至少分成四層入口層FastAPI 路由、鏈路層Redis 讀寫、聚合層消費(fèi)者任務(wù)、推送層WebSocket 管理器。目錄結(jié)構(gòu)如下event_analyzer/ ├── app/ │ ├── main.py │ ├── api/ │ │ └── events.py │ ├── services/ │ │ ├── stream.py │ │ ├── aggregator.py │ │ ├── lifecycle.py │ │ └── pusher.py │ ├── models/ │ │ └── schemas.py │ └── config.py ├── scripts/ │ └── run_worker.py ├── tests/ └── docker-compose.yml這種分層不是為了好看是為了定位問題。聚合邏輯如果和 HTTP 路由混在一起排障時(shí)你會(huì)先懷疑請(qǐng)求有問題再懷疑聚合有問題最后才發(fā)現(xiàn)是兩邊的狀態(tài)錯(cuò)位。分干凈以后每一層都能獨(dú)立測(cè)試也方便以后替換組件。3.2 上報(bào)接口格式校驗(yàn)與冪等設(shè)計(jì)事件上報(bào)是整個(gè)系統(tǒng)的入口也是數(shù)據(jù)質(zhì)量的第一道防線。這里用 Pydantic 模型做格式校驗(yàn)事件類型走白名單不允許隨便傳。# app/models/schemas.py from typing import Optional from datetime import datetime from pydantic import BaseModel, Field, field_validator class EventPayload(BaseModel): event_type: str Field(..., min_length1, max_length50) user_id: str Field(..., min_length1, max_length64) page: str Field(unknown, max_length255) client_time: Optional[datetime] None payload: dict Field(default_factorydict) field_validator(event_type) classmethod def check_event_type(cls, v): allowed {click, exposure, login, submit_order, pay} if v not in allowed: raise ValueError(funsupported event type: {v}) return v入口校驗(yàn)不只是為了攔截惡意調(diào)用更是為了攔截業(yè)務(wù)方自己的不規(guī)范調(diào)用。我在做這個(gè)項(xiàng)目時(shí)曾經(jīng)因?yàn)闆]限制事件類型一個(gè)測(cè)試腳本把各種亂寫的值灌進(jìn)流水線聚合數(shù)據(jù)里全是垃圾花了半天才發(fā)現(xiàn)是入口沒有白名單。這個(gè)教訓(xùn)很直接寧可讓上報(bào)接口多一步校驗(yàn)也不要在聚合端浪費(fèi)時(shí)間清洗。上報(bào)路由的代碼很薄只負(fù)責(zé)生成事件 ID、補(bǔ)充服務(wù)端時(shí)間、寫入 Stream# app/api/events.py import uuid from datetime import datetime, timezone from fastapi import APIRouter from app.models.schemas import EventPayload from app.services.stream import event_stream router APIRouter(prefix/api/v1, tags[events]) router.post(/events) async def report_event(evt: EventPayload): event_id f{datetime.now(timezone.utc).strftime(%Y%m%d%H%M%S)}-{uuid.uuid4().hex[:8]} data { event_id: event_id, event_type: evt.event_type, user_id: evt.user_id, page: evt.page or unknown, client_time: evt.client_time.isoformat() if evt.client_time else datetime.now(timezone.utc).isoformat(), payload: evt.payload, } event_stream.add(data) return {code: 0, message: ok, event_id: event_id}event_id由服務(wù)端生成而不是信任客戶端傳入這樣客戶端重試時(shí)拿到的永遠(yuǎn)是同一個(gè) ID后文消費(fèi)端會(huì)利用這個(gè) ID 做冪等判斷避免重復(fù)計(jì)數(shù)。3.3 聚合邏輯PV/UV 與漏斗計(jì)算聚合是系統(tǒng)里最核心的邏輯我把它放在一個(gè)獨(dú)立 worker 里。這個(gè) worker 從 Redis Streams 批量取事件更新內(nèi)存計(jì)數(shù)器和 Redis 集合。# app/services/aggregator.py from collections import Counter from datetime import datetime, timezone from app.services.stream import stream_client as redis # 內(nèi)存統(tǒng)計(jì)按「小時(shí):事件類型:頁面」精確計(jì)數(shù) minute_counter Counter() funnel_prefix funnel def process_event(raw: dict): hour raw[client_time][:13] key f{hour}:{raw[event_type]}:{raw[page]} minute_counter[key] 1 uid raw[user_id] etype raw[event_type] fkey f{funnel_prefix}:{hour} # 漏斗登錄 - 提交訂單 - 支付 if etype login: redis.sadd(f{fkey}:step1, uid) elif etype submit_order: if redis.sismember(f{fkey}:step1, uid): redis.sadd(f{fkey}:step2, uid) elif etype pay: if redis.sismember(f{fkey}:step2, uid): redis.sadd(f{fkey}:step3, uid)這套漏斗實(shí)現(xiàn)用的是 Redis Set 的成員判斷先登錄的用戶進(jìn)入 step1 集合提交訂單前檢查是否在 step1 中支付前檢查是否在 step2 中。實(shí)現(xiàn)很樸素但足夠支撐第一版的轉(zhuǎn)化趨勢(shì)判斷。為什么漏斗集合必須放在 Redis 而不是進(jìn)程內(nèi)因?yàn)榫酆?worker 可能不止一個(gè)進(jìn)程如果每個(gè)進(jìn)程各自維護(hù)一套集合分布式部署時(shí)數(shù)據(jù)就會(huì)不一致。Redis 的SADD和SISMEMBER都是原子操作天然支持多進(jìn)程共享。這里的取舍是原則上是能放內(nèi)存就放內(nèi)存但數(shù)據(jù)和狀態(tài)需要跨進(jìn)程共享時(shí)就得上 Redis 這類獨(dú)立存儲(chǔ)。消費(fèi) worker 的循環(huán)長(zhǎng)這樣# scripts/run_worker.py import time from app.services.stream import stream_client as redis from app.services.aggregator import process_event STREAM_KEY event_stream:raw GROUP_NAME aggregator_group CONSUMER_NAME consumer_1 def ensure_group(): try: redis.xgroup_create(STREAM_KEY, GROUP_NAME, id0, mkstreamTrue) except Exception: pass def consume(): ensure_group() while True: results redis.xreadgroup( GROUP_NAME, CONSUMER_NAME, {STREAM_KEY: }, count200, block2000 ) if not results: continue for _, event_list in results: for event_id, fields in event_list: process_event(fields) redis.xack(STREAM_KEY, GROUP_NAME, event_id) if __name__ __main__: consume()這段代碼里要注意XACK的位置必須在事件處理成功之后再確認(rèn)。如果先確認(rèn)后處理消費(fèi)端在兩者之間崩潰事件會(huì)悄悄丟失。反過來如果處理完不確認(rèn)這個(gè)事件在 Redis 里會(huì)一直被消費(fèi)組記住重啟后會(huì)重復(fù)消費(fèi)。配合冪等設(shè)計(jì)才能保證數(shù)據(jù)準(zhǔn)確。3.4 WebSocket 實(shí)時(shí)推送從后端到瀏覽器推送層的核心是一個(gè)連接管理器。它維護(hù)當(dāng)前所有活躍的 WebSocket 連接聚合結(jié)果變化時(shí)向所有連接廣播。# app/services/pusher.py from fastapi import WebSocket class ConnectionManager: def __init__(self): self.active_connections [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) async def broadcast(self, message: dict): if not self.active_connections: return stale [] for conn in self.active_connections: try: await conn.send_json(message) except Exception: stale.append(conn) for conn in stale: self.active_connections.remove(conn) manager ConnectionManager()推送端最容易犯的錯(cuò)誤是只加連接不清理斷開的連接。如果某個(gè)瀏覽器標(biāo)簽頁一直掛著但網(wǎng)絡(luò)已經(jīng)斷掉send_json會(huì)持續(xù)拋異常積少成多以后整個(gè)廣播循環(huán)被拖住。所以我在每次廣播時(shí)都會(huì)順手把所有異常連接從列表里移除。實(shí)測(cè)下來這個(gè)處理比單純依賴WebSocketDisconnect更可靠因?yàn)閿嚅_時(shí)機(jī)不一定能準(zhǔn)確觸發(fā)異常事件。對(duì)應(yīng)的 WebSocket 路由# app/main.py from fastapi import FastAPI, WebSocket, WebSocketDisconnect from app.services.pusher import manager app FastAPI() app.websocket(/ws/dashboard) async def dashboard_socket(websocket: WebSocket): await manager.connect(websocket) try: while True: await websocket.receive_text() except WebSocketDisconnect: manager.active_connections.remove(websocket)前端最簡(jiǎn)單也要保留一分鐘的最近消息時(shí)間戳方便確認(rèn)連接是活的const ws new WebSocket(ws://${location.host}/ws/dashboard); ws.onmessage (event) { const data JSON.parse(event.data); if (data.type overview) renderOverview(data.data); if (data.type funnel) renderFunnel(data.data); lastMessageAt Date.now(); };上面的代碼只是骨架完整看板還要處理重連邏輯。WebSocket 會(huì)因?yàn)榇砜臻e超時(shí)、網(wǎng)絡(luò)切換等原因斷開前端需要在onclose里加指數(shù)退避重連否則看板會(huì)永遠(yuǎn)停留在斷開狀態(tài)。3.5 部署與驗(yàn)證docker-compose 一次拉起來整套系統(tǒng)用 docker-compose 啟動(dòng)最省事一個(gè) Redis、一個(gè) API 服務(wù)、一個(gè) worker 服務(wù)。# docker-compose.yml version: 3.8 services: redis: image: redis:7.0-alpine ports: - 6379:6379 api: build: . depends_on: - redis environment: - REDIS_HOSTredis ports: - 8000:8000 command: uvicorn app.main:app --host 0.0.0.0 --port 8000 worker: build: . depends_on: - redis environment: - REDIS_HOSTredis command: python scripts/run_worker.py壓測(cè)時(shí)我用wrk對(duì)/api/v1/events按 3000 QPS 跑了幾分鐘API 進(jìn)程占用大約 60%Redis 基本無壓力。前端儀表盤通過 WebSocket 收到的新數(shù)據(jù)與實(shí)時(shí)寫入 Redis 的計(jì)數(shù)一致說明整條鏈路能閉合。提示在真實(shí)環(huán)境做壓測(cè)一定不要把壓測(cè)流量和線上流量混在同一個(gè) Redis 實(shí)例否則數(shù)據(jù)會(huì)相互干擾你很難判斷異常來自測(cè)試還是業(yè)務(wù)。4. 常見問題與排查技巧實(shí)錄再穩(wěn)定的小系統(tǒng)也會(huì)在新版本上線后突然變得別扭。下面這幾個(gè)問題是我在實(shí)際調(diào)試中真實(shí)遇到過的記錄下來比背一百個(gè) API 都有用。4.1 Redis 內(nèi)存不斷上漲從消費(fèi)端找原因跑了一段時(shí)間后我注意到 Redis 內(nèi)存穩(wěn)步上漲。最開始懷疑是 Stream 的原始數(shù)據(jù)沒清掉后來檢查發(fā)現(xiàn)是聚合用的漏斗 key 沒設(shè)過期。原因是我把漏斗 key 設(shè)成了funnel:{小時(shí)}但沒有給它們?cè)O(shè)置 TTL。一天 24 個(gè) key 看起來不多可是每個(gè) key 里的用戶 ID 會(huì)持續(xù)累積。一個(gè)月下來全是無用數(shù)據(jù)內(nèi)存就這樣被吃掉了。解決辦法是在聚合 worker 里加一個(gè)定時(shí)清理任務(wù)把所有超過 48 小時(shí)沒更新的漏斗 key 直接刪除。聚合結(jié)果屬于熱數(shù)據(jù)保留兩天足夠支撐看板回看再舊的歷史數(shù)據(jù)應(yīng)該走持久層不該放在 Redis 里。4.2 事件丟失與重復(fù)Stream 的確認(rèn)和冪等是關(guān)鍵最頭疼的一類問題是前端看板顯示的數(shù)字和后臺(tái)數(shù)據(jù)庫不一致。排查后發(fā)現(xiàn)第一個(gè)原因是消費(fèi)者讀取 Stream 后沒有XACK導(dǎo)致崩潰重啟后同一批事件被重復(fù)消費(fèi)消費(fèi)端又沒做冪等所以計(jì)數(shù)多了。后來我把消費(fèi)邏輯改成了先XREADGROUP批量取出處理完成后XACK再對(duì)每個(gè)event_id做冪等標(biāo)記。具體做法是在處理事件前執(zhí)行一個(gè)SETNX只有第一次成功設(shè)置時(shí)才真的參與計(jì)算否則跳過。這個(gè)操作為每一條事件多花了一次 Redis 請(qǐng)求但換來了數(shù)據(jù)的最終一致性。4.3 實(shí)時(shí)推送延遲變大別把阻塞操作放進(jìn) IO 循環(huán)有段時(shí)間儀表盤數(shù)據(jù)要延遲十幾秒才刷新最初懷疑 WebSocket 連接斷了后來發(fā)現(xiàn)是聚合 worker 在處理事件時(shí)調(diào)用了一個(gè)同步的外部配置接口。Python 協(xié)程調(diào)度被這個(gè)同步請(qǐng)求卡住整個(gè) worker 的處理速度被拖慢。線上排查這類問題最直接的方式是看 worker 的日志時(shí)間戳如果處理日志中間有很長(zhǎng)的空洞基本可以斷定有阻塞調(diào)用。實(shí)時(shí)鏈路里盡量不要混用同步阻塞庫。如果一定要調(diào)用外部服務(wù)用httpx.AsyncClient或把配置緩存到 Redis短時(shí)間內(nèi)不重復(fù)獲取。實(shí)時(shí)系統(tǒng)最怕的就是某個(gè)不起眼的阻塞點(diǎn)拖住整個(gè)消費(fèi)進(jìn)度。4.4 數(shù)據(jù)與圖表對(duì)不上窗口對(duì)齊的坑曾有一次漏斗轉(zhuǎn)化率和前端曲線相互矛盾明明系統(tǒng)顯示的是 10% 的轉(zhuǎn)化率手工數(shù)事件卻發(fā)現(xiàn)對(duì)不上。原因是聚合時(shí)我用了消費(fèi)端本地時(shí)間做小時(shí)分桶但上報(bào)事件本身帶有client_time在跨時(shí)區(qū)場(chǎng)景下兩者會(huì)出現(xiàn)錯(cuò)位。后來統(tǒng)一改成只按client_time分桶服務(wù)端和前端展示用同一個(gè)時(shí)間維度數(shù)據(jù)才真正對(duì)齊。窗口問題在實(shí)時(shí)分析里很容易被忽略。建議從第一天就統(tǒng)一時(shí)間語義事件時(shí)間還是消費(fèi)時(shí)間。趨勢(shì)看板通常應(yīng)該用事件時(shí)間因?yàn)樗咏鼧I(yè)務(wù)真實(shí)發(fā)生時(shí)間只有關(guān)心系統(tǒng)吞吐時(shí)才用消費(fèi)時(shí)間。4.5 常見問題速查表癥狀可能原因處理建議Redis 內(nèi)存持續(xù)上漲聚合 key 未設(shè)過期或 Stream 未裁剪給 key 加 TTL設(shè)置 MAXLEN看板數(shù)據(jù)偏大重復(fù)消費(fèi)導(dǎo)致重復(fù)計(jì)數(shù)XACK 加 SETNX 冪等推送延遲高同步阻塞調(diào)用 / 連接池滿改異步調(diào)用提高連接池上限漏斗數(shù)據(jù)對(duì)不上時(shí)間窗口不統(tǒng)一統(tǒng)一用 client_time 分桶WebSocket 頻繁掉線代理空閑超時(shí)加心跳消息縮短 ping 間隔5. 后續(xù)擴(kuò)展思路這套骨架還能長(zhǎng)成什么做完第一版以后很容易產(chǎn)生要不直接上 ClickHouse的想法。我的建議是別急先把架構(gòu)能延伸的方向理清楚再?zèng)Q定下一步往哪兒長(zhǎng)。5.1 從指標(biāo)分析到異常檢測(cè)當(dāng) PV/UV 和漏斗數(shù)據(jù)穩(wěn)定運(yùn)行后自然會(huì)產(chǎn)生一個(gè)需求系統(tǒng)能不能自動(dòng)發(fā)現(xiàn)異常。比如某小時(shí) PV 掉了 40%或者某個(gè)頁面的點(diǎn)擊率突然飆升。這種功能可以從簡(jiǎn)單的規(guī)則開始每個(gè)小時(shí)跑一次統(tǒng)計(jì)把當(dāng)前指標(biāo)和過去 14 天的同期均值做差超過 3 倍標(biāo)準(zhǔn)差就發(fā)出告警。我試過先用最簡(jiǎn)單的方式實(shí)現(xiàn)不引入專門的學(xué)習(xí)庫只用numpy算均值和標(biāo)準(zhǔn)差把告警規(guī)則寫進(jìn)配置文件。這樣做的最大好處是透明可解釋出了問題你能直接定位是哪個(gè)指標(biāo)觸發(fā)的告警而不是像黑盒模型那樣無從下手。5.2 多租戶隔離把 project_id 寫進(jìn)每個(gè)事件如果這個(gè)系統(tǒng)要服務(wù)多個(gè)業(yè)務(wù)方最簡(jiǎn)單的做法不是在數(shù)據(jù)庫層做復(fù)雜隔離而是在每個(gè)事件里加一個(gè)project_id字段。所有 Redis key 都以project_id作前綴聚合邏輯不變只是查詢維度多了一個(gè)過濾條件。需要注意的坑是 WebSocket 推送不能再把所有項(xiàng)目的數(shù)據(jù)全量推給每個(gè)連接。前端連接時(shí)帶上project_id后端維護(hù)一個(gè)按項(xiàng)目區(qū)分的連接映射這樣才能避免不同業(yè)務(wù)方的數(shù)據(jù)串臺(tái)。這個(gè)問題我在測(cè)試期踩過當(dāng)時(shí)一個(gè)測(cè)試項(xiàng)目的曝光數(shù)據(jù)直接出現(xiàn)在另一個(gè)項(xiàng)目的看板上排查了很久才意識(shí)到是連接映射沒加過濾。5.3 冷熱數(shù)據(jù)分層從 Redis 到數(shù)據(jù)倉庫Redis 保存熱數(shù)據(jù)沒問題但要長(zhǎng)期保存歷史趨勢(shì)、做深層次回溯它就不合適了。建議每天凌晨定時(shí)把前一天的聚合結(jié)果從 Redis 導(dǎo)出到 PostgreSQL并為按天建分區(qū)表。這樣歷史查詢走持久層實(shí)時(shí)看板走 Redis兩個(gè)鏈路互不干擾。如果事件量之后漲到千萬級(jí)甚至更高可以在 PostgreSQL 前面加一個(gè)輕量 OLAP 引擎做維度查詢。擴(kuò)展的關(guān)鍵仍然是要先把熱鏈路跑穩(wěn)再讓冷數(shù)據(jù)落地而不是一上來就鋪一堆基礎(chǔ)設(shè)施。最后分享一個(gè)我認(rèn)為最值得記住的經(jīng)驗(yàn)?zāi)玫筋愃苧ea這種看不出含義的代號(hào)時(shí)不要把時(shí)間全花在猜測(cè)語義上。先找一個(gè)能獨(dú)立形成閉環(huán)的落地方向把上報(bào)、聚合、推送三條鏈路串起來讓數(shù)據(jù)真正流動(dòng)起來之后所有疑問都會(huì)在真實(shí)運(yùn)行中自行顯形。模糊的稱呼只是起點(diǎn)清晰的技術(shù)邊界才是你能掌控的東西。