数据源接入：常见数据源类型与实时性保障

做实时风险仪表盘，第一步就是搞定数据源。说白了，你得先知道数据从哪来，怎么拿，拿得快不快。

我刚开始做这类项目时，踩过一个坑——数据源没选对，后面所有优化都白搭。所以这一章，咱们把数据源这件事彻底聊透。

常见数据源类型

实时仪表盘的数据源，我归纳下来就三大类：API、数据库、WebSocket。每种都有它的脾气。

1. API（HTTP/HTTPS 接口）

这是最常用的方式。你的系统通过 HTTP 请求，从外部服务拉取数据。

• RESTful API：最常见，基于请求-响应模式。适合低频轮询，比如每 5 秒查一次。
• GraphQL：可以精确指定要哪些字段，减少数据传输量。我曾在风控项目里用它，效果不错。

// 一个简单的 API 轮询示例（JavaScript）
async function fetchRiskData() {
  const response = await fetch('https://api.risk-dashboard.com/v1/events');
  const data = await response.json();
  updateDashboard(data);
}

// 每 5 秒拉一次
setInterval(fetchRiskData, 5000);

2. 数据库（MySQL、PostgreSQL、ClickHouse 等）

直接从数据库查数据，适合内部系统。但要注意，频繁查询会压垮数据库。

• 关系型数据库：MySQL、PostgreSQL。适合结构化数据，但高并发查询要小心。
• 时序数据库：InfluxDB、TimescaleDB。专门为时间序列数据优化，我强烈推荐用于风险指标。
• 列式存储：ClickHouse。分析查询极快，适合聚合计算。

-- 一个简单的查询示例（PostgreSQL）
SELECT 
  event_type,
  COUNT(*) as event_count,
  AVG(risk_score) as avg_risk
FROM risk_events
WHERE created_at > NOW() - INTERVAL '5 minutes'
GROUP BY event_type;

3. WebSocket（全双工通信）

这是实时性最强的方案。服务端可以主动推送数据，客户端不用轮询。

• 原生 WebSocket：浏览器原生支持，轻量高效。
• Socket.IO：封装了重连、心跳等机制，更稳定。
• SSE（Server-Sent Events）：单向推送，适合服务端到客户端的场景。

// WebSocket 客户端示例
const ws = new WebSocket('wss://risk-stream.example.com');

ws.onmessage = function(event) {
  const riskEvent = JSON.parse(event.data);
  // 直接更新仪表盘
  updateRealTimeChart(riskEvent);
};

ws.onclose = function() {
  console.log('连接断开，尝试重连...');
  // 重连逻辑
};

数据采集策略

数据源选好了，接下来就是怎么采。策略不对，实时性就是空话。

1. 轮询（Polling）

客户端定时发起请求。简单粗暴，但效率低。

• 固定间隔轮询：比如每 2 秒查一次。适合数据变化不频繁的场景。
• 自适应轮询：根据数据变化频率动态调整间隔。数据变化快就缩短间隔，反之拉长。

2. 推送（Push）

服务端主动推送数据。实时性最好，资源消耗也低。

• WebSocket 推送：全双工，双向通信。
• 消息队列：Kafka、RabbitMQ。适合高吞吐场景，数据先入队，再消费。

// Kafka 消费者示例（Node.js）
const { Kafka } = require('kafkajs');

const kafka = new Kafka({ clientId: 'risk-dashboard', brokers: ['localhost:9092'] });
const consumer = kafka.consumer({ groupId: 'risk-group' });

await consumer.connect();
await consumer.subscribe({ topic: 'risk-events', fromBeginning: true });

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const riskData = JSON.parse(message.value.toString());
    // 推送到前端
    broadcastToClients(riskData);
  },
});

3. 增量采集

只采集变化的数据，而不是全量拉取。能大幅减少数据传输量。

• 时间戳增量：记录上次采集时间，只取之后的数据。
• 版本号增量：每条数据带版本号，只取更新的。
• CDC（Change Data Capture）：监听数据库的变更日志，实时捕获变化。

实时性保障

实时性不是玄学，是工程。我总结了几条硬指标和落地方法。

1. 延迟指标

级别	延迟要求	适用场景
硬实时	< 100ms	交易风控、高频告警
准实时	100ms - 1s	监控大盘、运营看板
近实时	1s - 10s	报表分析、趋势展示

2. 保障手段

• 数据缓冲：用内存队列或 Redis 做缓冲，防止突发流量打垮后端。
• 异步处理：数据采集和仪表盘渲染解耦，互不影响。
• 连接池：数据库和 API 都用连接池，减少连接建立开销。
• 数据压缩：传输时用 gzip 压缩，减少网络延迟。

知识体系总览

下面这张图，把数据源接入的核心逻辑串起来了。你一看就明白。