“昨天UV多少？”——隔天的数据谁都能查。

“现在有多少人在线？”——这才是实时数据的价值。

这篇文章，我会教你设计一个毫秒级延迟的实时数据统计系统。

---

一、为什么需要实时数据？

1.1 批处理 vs 实时处理

维度	批处理（T+1）	实时处理（秒级）
延迟	次日	秒级/毫秒级
价值	分析决策	实时监控/运营
场景	日报、周报	活动监控、异常告警
成本	低	高
复杂度	低	高

---

1.2 实时数据的典型场景

场景	需求	延迟要求
活动监控	裂变活动进行中，实时看参与人数	秒级
异常告警	UV突降50%，立即通知	毫秒级
实时推荐	用户正在浏览，推荐相关内容	毫秒级
在线人数	当前在线用户数	秒级
A/B测试	实时看实验组数据	分钟级

---

二、实时系统架构

2.1 整体架构

code复制

┌──────────────────────────────────────────────────────────┐
│                       实时展示层                          │
│              WebSocket + Vue/React                        │
├──────────────────────────────────────────────────────────┤
│                       实时计算层                          │
│           Flink / 自研实时计算引擎                         │
├───────────────┬──────────────────┬───────────────────────┤
│   实时存储     │    离线存储       │     缓存              │
│  Redis/TDengine│  ClickHouse     │    Redis              │
├───────────────┴──────────────────┴───────────────────────┤
│                       消息队列                            │
│                    Kafka / Pulsar                         │
├──────────────────────────────────────────────────────────┤
│                       采集层                              │
│              前端SDK → 上报网关                            │
└──────────────────────────────────────────────────────────┘

---

2.2 数据流

code复制

前端埋点 → 上报网关 → Kafka → Flink → Redis（实时） + ClickHouse（落盘）
                                    ↓
                              WebSocket → 看板

---

三、时序数据库选型

数据库	写入速度	查询速度	时序支持	适合实时
MySQL	低	中	差	❌
ClickHouse	高（批量）	高（批量）	中	❌
Redis	极高	极高	差（需自己设计）	✅
InfluxDB	高	高	好	✅
TDengine	极高	极高	好	✅

推荐方案：Redis + TDengine

用途	数据库	理由
当前状态	Redis	读写极快，适合存当前在线人数
时序数据	TDengine	专为时序数据设计，写入极快
离线分析	ClickHouse	批量分析，留存/漏斗/归因

---

四、实时计算设计

4.1 实时指标定义

指标	定义	计算方式	精度
实时在线	最近5分钟内活跃的用户数	Redis Set，5分钟过期	精确
分钟UV	每分钟独立用户数	滑动窗口去重	精确
实时PV	每分钟页面浏览量	滑动窗口计数	精确
实时转化率	实时支付人数/实时访问人数	双窗口比率	精确

---

4.2 Redis实时状态设计

javascript复制

class RealtimeState {
  constructor(redis) { this.redis = redis; }

  async userActive(userId) {
    const now = Date.now();
    const pipe = this.redis.pipeline();

    // 1. 在线用户集合（5分钟过期）
    pipe.sadd('online:users', userId);
    pipe.expire('online:users', 300);

    // 2. 分钟UV（HyperLogLog）
    const minuteKey = `uv:min:${this.getMinuteKey(now)}`;
    pipe.pfadd(minuteKey, userId);
    pipe.expire(minuteKey, 3600);

    // 3. 分钟PV
    const pvKey = `pv:min:${this.getMinuteKey(now)}`;
    pipe.incr(pvKey);
    pipe.expire(pvKey, 3600);

    // 4. 在线时长追踪（Sorted Set）
    pipe.zadd('online:timeline', now, userId);
    pipe.zremrangebyscore('online:timeline', '-inf', now - 300000);

    await pipe.exec();
  }

  async getOnlineCount() { return await this.redis.scard('online:users'); }

  async getMinuteUV(minutesAgo = 0) {
    const key = `uv:min:${this.getMinuteKey(Date.now() - minutesAgo * 60000)}`;
    return await this.redis.pfcount(key);
  }

  async getMinuteUVTrend(minutes = 30) {
    const result = [];
    for (let i = minutes - 1; i >= 0; i--) {
      const uv = await this.getMinuteUV(i);
      result.push({ time: this.getMinuteKey(Date.now() - i * 60000), uv });
    }
    return result;
  }

  getMinuteKey(ts) {
    const d = new Date(ts);
    return `${d.getFullYear()}${String(d.getMonth() + 1).padStart(2, '0')}${String(d.getDate()).padStart(2, '0')}${String(d.getHours()).padStart(2, '0')}${String(d.getMinutes()).padStart(2, '0')}`;
  }
}

---

4.3 TDengine时序存储设计

sql复制

CREATE DATABASE IF NOT EXISTS miniprogram_stats PRECISION 'ms' KEEP 3650 BUFFER 96 PAGES 256;
USE miniprogram_stats;

-- 事件时序超级表
CREATE STABLE IF NOT EXISTS events (
  ts            TIMESTAMP,
  event_name    NCHAR(50),
  session_id    NCHAR(50),
  user_id       NCHAR(64),
  properties    NCHAR(1024)
) TAGS (app_id NCHAR(32), platform NCHAR(16), version NCHAR(16));

-- 分钟汇总超级表
CREATE STABLE IF NOT EXISTS minute_summary (
  ts            TIMESTAMP,
  uv            INT,
  pv            INT,
  new_user      INT,
  online_count  INT
) TAGS (app_id NCHAR(32));

-- 查询：最近30分钟UV趋势
SELECT ts, uv, pv, online_count
FROM minute_summary
WHERE app_id = 'your_app_id' AND ts >= now - 30m
ORDER BY ts;

-- 查询：今日每小时UV
SELECT _wstart AS hour, sum(uv) AS uv, sum(pv) AS pv
FROM minute_summary
WHERE app_id = 'your_app_id' AND ts >= today()
INTERVAL(1h) ORDER BY hour;

---

4.4 Flink实时计算

java复制

public class RealtimeStatsJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        KafkaSource<TrackEvent> source = KafkaSource.<TrackEvent>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("track-events")
            .setGroupId("flink-realtime-stats")
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new TrackEventDeserializer())
            .build();

        DataStream<TrackEvent> events = env.fromSource(
            source, WatermarkStrategy.noWatermarks(), "kafka-source"
        );

        // 1. 实时在线人数（5分钟滑动窗口，10秒滑动）
        events.keyBy(e -> e.getAppId())
            .window(SlidingProcessingTimeWindows.of(Time.minutes(5), Time.seconds(10)))
            .aggregate(new OnlineCountAgg())
            .addSink(new RedisSink<>("online:count"));

        // 2. 分钟级UV/PV（1分钟滚动窗口）
        events.keyBy(e -> e.getAppId())
            .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
            .aggregate(new MinuteStatsAgg())
            .addSink(new TDengineSink<>("minute_summary"));

        // 3. 异常检测
        events.keyBy(e -> e.getAppId())
            .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
            .aggregate(new AnomalyDetectAgg())
            .filter(alert -> alert != null)
            .addSink(new AlertSink());

        env.execute("Realtime Stats Job");
    }
}

---

五、WebSocket推送设计

5.1 服务端

javascript复制

const WebSocket = require('ws');
const Redis = require('ioredis');

const wss = new WebSocket.Server({ port: 8081 });
const redis = new Redis();
const clients = new Map();

wss.on('connection', (ws) => {
  ws.on('message', (message) => {
    const data = JSON.parse(message);
    if (data.type === 'subscribe') clients.set(ws, { appId: data.appId });
  });
  ws.on('close', () => clients.delete(ws));
});

// 每10秒推送
setInterval(async () => {
  for (const [ws, config] of clients) {
    if (ws.readyState !== WebSocket.OPEN) continue;
    const data = await getRealtimeData(config.appId);
    ws.send(JSON.stringify({ type: 'realtime_update', data, timestamp: Date.now() }));
  }
}, 10000);

5.2 前端

vue复制

<template>
  <div class="realtime-dashboard">
    <div class="realtime-card online">
      <div class="label">当前在线</div>
      <div class="value pulse">{{ data.online_count || 0 }}</div>
    </div>
    <div ref="uvChart" style="height: 200px;"></div>
  </div>
</template>

<script setup>
import { ref, onMounted, onUnmounted } from 'vue';
import * as echarts from 'echarts';

const data = ref({});
let ws = null;

onMounted(() => {
  ws = new WebSocket('ws://localhost:8081');
  ws.onopen = () => ws.send(JSON.stringify({ type: 'subscribe', appId: 'your_app_id' }));
  ws.onmessage = (event) => {
    const msg = JSON.parse(event.data);
    if (msg.type === 'realtime_update') data.value = msg.data;
  };
});
onUnmounted(() => { if (ws) ws.close(); });
</script>

---

六、异常告警设计

6.1 告警规则

规则	条件	级别	通知方式
UV突降	当前分钟UV < 上分钟UV × 50%	P0	企业微信 + 短信
PV异常	当前PV > 日常PV × 3	P1	企业微信
在线归零	在线人数 = 0 超过5分钟	P0	企业微信 + 电话
上报延迟	Kafka消费延迟 > 5分钟	P1	企业微信
错误率上升	错误率 > 5%	P2	邮件

---

6.2 告警服务

javascript复制

class AlertService {
  constructor(options) {
    this.redis = options.redis;
    this.alertCallback = options.alertCallback;
    this.rules = options.rules || [];
    this.cooldown = new Map();
  }

  async check(appId, currentData) {
    for (const rule of this.rules) {
      const shouldAlert = await rule.evaluate(appId, currentData, this.redis);
      if (shouldAlert) {
        const cooldownKey = `${appId}:${rule.name}`;
        if (this.cooldown.get(cooldownKey) > Date.now()) continue;
        await this.alertCallback({
          appId, ruleName: rule.name, level: rule.level,
          message: shouldAlert.message, data: currentData, timestamp: Date.now()
        });
        this.cooldown.set(cooldownKey, Date.now() + rule.cooldownMs);
      }
    }
  }
}

---

七、性能优化

7.1 写入优化

优化	方案	效果
批量写入	Kafka攒批，每1000条或每秒写入	吞吐量提升10倍
异步写入	写入和查询分离	查询不受写入影响
预聚合	Flink窗口聚合后写入	数据量减少90%
冷热分离	热→Redis，温→TDengine，冷→ClickHouse	查询性能提升5倍

7.2 查询优化

优化	方案	效果
缓存	Redis缓存，10秒过期	秒级→毫秒级
降采样	长范围按分钟/小时/天聚合	数据量减少100倍
预计算	提前计算常用指标	复杂查询变简单

7.3 推送优化

优化	方案	效果
增量推送	只推送变化数据	数据量减少80%
按需推送	客户端订阅	减少无用推送
压缩	permessage-deflate	数据量减少60%

---

八、方案选型

规模	方案	月成本
0-1万UV	不需要实时，微信后台T+1够用	0
1万-10万UV	轻量：Redis + 定时任务	~500元
10万-100万UV	标准：Redis + TDengine + WebSocket	~3000元
100万UV+	重量：全链路实时 Flink + Kafka	~15000元

---

写在最后

实时数据的核心价值不在于"看"，而在于"做"——实时发现问题，实时做出决策。

但切记：实时系统的复杂度和成本远高于离线系统。在业务不需要的时候，不要过早引入实时架构。