基于 SQL Server CDC 与 SSE 构建 Ant Design 实时数据网格的架构实践

全栈开发

文章字数: 3.8k

阅读时长: 17 分

项目需求很明确：一个后台监控面板，需要实时、无延迟地反映核心业务表（dbo.Orders）的数据变更。用户操作员必须在数据写入数据库的瞬间，就能在前端表格上看到新增、修改或删除的记录，以便进行即时干预。最初的方案是前端轮询，但每秒一次的轮询在高并发时对数据库和网络造成了无法接受的压力，且存在数秒的延迟。WebSocket方案被提出，但对于这种服务器到客户端的单向数据流场景，其双向通信协议显得过于笨重，增加了不必要的复杂度和资源开销。

我们需要一个更轻量、更直接的方案。技术栈选型最终锁定在一个组合上：SQL Server 的变更数据捕获 (Change Data Capture, CDC) 作为数据源头，Node.js 后端通过 Server-Sent Events (SSE) 将变更推送至前端，React 结合 Ant Design 的 Table 组件负责展示，并使用 Valtio 进行精细化的状态管理，以避免大规模数据变更时引发的性能雪崩。

第一步：在数据库层面捕获变更

一切的起点是数据源。与其在应用层做文章，不如直接让数据库告诉我们发生了什么。SQL Server 内置的 CDC 功能正是为此而生。它通过读取事务日志，异步地将 DML 操作（INSERT, UPDATE, DELETE）记录到系统创建的变更表中。

首先，必须在数据库级别启用 CDC：

-- 确保 SQL Server Agent 正在运行，CDC 依赖它来执行捕获和清理作业
USE YourDatabase;
GO
EXEC sys.sp_cdc_enable_db;
GO

然后，针对需要监控的 Orders 表启用 CDC。假设 Orders 表结构如下：

CREATE TABLE dbo.Orders (
    OrderID INT PRIMARY KEY IDENTITY,
    CustomerID INT NOT NULL,
    OrderDate DATETIME2 DEFAULT GETDATE(),
    Status NVARCHAR(50) NOT NULL,
    TotalAmount DECIMAL(18, 2) NOT NULL
);

启用 CDC 的命令：

EXEC sys.sp_cdc_enable_table
    @source_schema = N'dbo',
    @source_name   = N'Orders',
    @role_name     = NULL, -- 使用 NULL 表示只有 sysadmin 和 db_owner 成员可以访问变更数据
    @supports_net_changes = 1; -- 允许查询净变化，虽然本次我们不用，但这是个好习惯
GO

执行后，SQL Server 会自动创建一系列对象，其中最核心的是变更表 cdc.dbo_Orders_CT。它的结构包含了原表的所有列，并附加了几个元数据列，如 __$start_lsn、__$end_lsn、__$seqval、__$operation 和 __$update_mask。__$operation 列是关键，它的值代表了操作类型：1=删除, 2=插入, 3=更新前的值, 4=更新后的值。

现在，任何对 dbo.Orders 的操作都会被捕获。例如，执行一个插入和更新：

INSERT INTO dbo.Orders (CustomerID, Status, TotalAmount) VALUES (101, 'Pending', 199.99);
UPDATE dbo.Orders SET Status = 'Shipped' WHERE OrderID = 1;

查询变更表 cdc.dbo_Orders_CT 就能看到这些操作的详细记录。我们的后端服务就是要消费这张表。

第二步：构建稳健的 SSE 推送服务

后端选用 Node.js 和 Express，因为它在处理高并发 I/O 密集型任务（如维持大量 SSE 连接）时表现出色。核心任务是：

维护一个SSE连接池。
定期轮询 cdc.dbo_Orders_CT 表获取增量变更。
将变更广播给所有连接的客户端。

项目结构与配置

/cdc-sse-service
|-- /src
|   |-- config.js         // 数据库及服务配置
|   |-- db.js             // SQL Server 连接模块
|   |-- sseService.js     // SSE 连接管理与事件推送
|   |-- cdcPoller.js      // CDC 变更数据轮询器
|   |-- index.js          // Express 应用入口
|-- package.json

src/config.js:

// src/config.js
require('dotenv').config();

// 在真实项目中，绝不应硬编码密码，使用环境变量或密钥管理服务
module.exports = {
  port: process.env.PORT || 3001,
  db: {
    server: process.env.DB_SERVER || 'localhost',
    port: parseInt(process.env.DB_PORT || '1433', 10),
    database: process.env.DB_DATABASE || 'YourDatabase',
    user: process.env.DB_USER || 'your_user',
    password: process.env.DB_PASSWORD || 'your_password',
    options: {
      encrypt: process.env.DB_ENCRYPT === 'true', // 对 Azure SQL 等需要
      trustServerCertificate: true // 本地开发时使用
    }
  },
  cdc: {
    pollingInterval: 2000, // 轮询间隔，单位毫秒
    tableName: 'dbo_Orders_CT'
  }
};

src/db.js:

// src/db.js
const sql = require('mssql');
const { db: dbConfig } = require('./config');

const poolPromise = new sql.ConnectionPool(dbConfig)
  .connect()
  .then(pool => {
    console.log('Connected to MSSQL');
    return pool;
  })
  .catch(err => {
    console.error('Database Connection Failed! Bad Config: ', err);
    process.exit(1);
  });

module.exports = {
  sql,
  poolPromise
};

SSE 服务核心实现

src/sseService.js 负责管理所有客户端连接。

// src/sseService.js
let clients = [];

function initializeSSE(req, res) {
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.flushHeaders(); // 立刻发送头信息

  const clientId = Date.now();
  const newClient = {
    id: clientId,
    res
  };
  clients.push(newClient);
  console.log(`Client ${clientId} connected. Total clients: ${clients.length}`);

  // 发送一个心跳包，防止连接因不活动而被代理或防火墙关闭
  const heartbeatInterval = setInterval(() => {
    res.write(': heartbeat\n\n');
  }, 15000);

  req.on('close', () => {
    clearInterval(heartbeatInterval);
    clients = clients.filter(client => client.id !== clientId);
    console.log(`Client ${clientId} disconnected. Total clients: ${clients.length}`);
  });
}

function broadcast(data) {
  if (!data || data.length === 0) {
    return;
  }
  
  // 这里的坑在于：必须为每个事件指定一个唯一的 ID。
  // 这使得客户端在断线重连后，可以通过 Last-Event-ID 头告诉服务器从哪个事件开始重新发送，实现事件的可靠传递。
  // 我们使用 LSN (Log Sequence Number) 作为天然的唯一 ID。
  const lastLsn = data[data.length - 1].__$start_lsn.toString('hex');
  
  const message = `id: ${lastLsn}\nevent: cdc_update\ndata: ${JSON.stringify(data)}\n\n`;

  clients.forEach(client => {
    try {
        client.res.write(message);
    } catch (error) {
        console.error(`Error sending to client ${client.id}:`, error);
        // 如果写入失败，可以考虑移除该客户端
    }
  });
}

module.exports = {
  initializeSSE,
  broadcast
};

CDC 轮询器

这是整个后端的引擎。它需要持久化记录上次查询到的 LSN (Log Sequence Number)，以确保不会漏掉或重复处理数据。

src/cdcPoller.js:

// src/cdcPoller.js
const { sql, poolPromise } = require('./db');
const { cdc: cdcConfig } = require('./config');
const { broadcast } = require('./sseService');

// 在生产环境中，这个状态应该持久化到 Redis 或数据库中，以防服务重启导致 LSN 丢失。
// 为简化示例，我们将其保存在内存中。
let lastLsn = null;

async function getInitialLsn() {
    try {
        const pool = await poolPromise;
        const result = await pool.request().query('SELECT sys.fn_cdc_get_max_lsn() as max_lsn');
        lastLsn = result.recordset[0].max_lsn;
        console.log(`Initial LSN set to: 0x${lastLsn.toString('hex')}`);
    } catch (error) {
        console.error('Failed to get initial max LSN:', error);
        // 如果无法获取初始 LSN，服务无法正常工作，应退出或重试
        process.exit(1);
    }
}

async function pollCdcChanges() {
    if (!lastLsn) {
        console.warn('Waiting for initial LSN...');
        return;
    }

    try {
        const pool = await poolPromise;
        const nextLsn = await pool.request().query('SELECT sys.fn_cdc_get_max_lsn() as max_lsn');
        const toLsn = nextLsn.recordset[0].max_lsn;

        // 如果没有新的变更，则不执行查询
        if (Buffer.compare(lastLsn, toLsn) >= 0) {
            return;
        }

        // 这里的查询是关键。我们使用 cdc.fn_cdc_get_all_changes_<capture_instance> 函数。
        // 它比直接查 _CT 表更安全、更高效。
        const query = `
            SELECT
                __$operation as operation,
                __$start_lsn,
                OrderID, CustomerID, OrderDate, Status, TotalAmount
            FROM cdc.fn_cdc_get_all_changes_dbo_Orders(
                @from_lsn, @to_lsn, N'all update old'
            )
            ORDER BY __$start_lsn, __$seqval;
        `;
        
        const result = await pool.request()
            .input('from_lsn', sql.VarBinary, sys.fn_cdc_increment_lsn(lastLsn))
            .input('to_lsn', sql.VarBinary, toLsn)
            .query(query);

        if (result.recordset.length > 0) {
            console.log(`Found ${result.recordset.length} changes.`);
            // 将二进制的 LSN 转换为十六进制字符串以便 JSON 序列化
            const changes = result.recordset.map(row => ({
                ...row,
                __$start_lsn: row.__$start_lsn.toString('hex')
            }));
            broadcast(changes);
            lastLsn = toLsn; // 更新 LSN
        }
    } catch (error) {
        console.error('Error polling CDC changes:', error);
        // 此处应添加更复杂的错误处理逻辑，如连接重试
    }
}

function startPolling() {
    getInitialLsn().then(() => {
        setInterval(pollCdcChanges, cdcConfig.pollingInterval);
        console.log(`CDC poller started. Interval: ${cdcConfig.pollingInterval}ms`);
    });
}

// 辅助函数，在 mssql 驱动中没有直接提供
const sys = {
    fn_cdc_increment_lsn: (lsn_buffer) => {
        // LSN 是一个10字节的二进制数，加1操作
        for (let i = lsn_buffer.length - 1; i >= 0; i--) {
            if (lsn_buffer[i] < 255) {
                lsn_buffer[i]++;
                return lsn_buffer;
            }
            lsn_buffer[i] = 0;
        }
        return lsn_buffer;
    }
}


module.exports = { startPolling };

最后，将它们在 index.js 中组合起来：

// src/index.js
const express = require('express');
const cors = require('cors');
const { port } = require('./config');
const { initializeSSE } = require('./sseService');
const { startPolling } = require('./cdcPoller');

const app = express();
app.use(cors());

app.get('/events', initializeSSE);

app.listen(port, () => {
  console.log(`SSE service listening on port ${port}`);
  startPolling();
});

至此，一个能够从 SQL Server CDC 获取变更并通过 SSE 推送的后端服务就绪了。

第三步：构建高性能的实时前端

前端的挑战在于如何高效地处理源源不断的数据流，并将其应用到 Ant Design 的 Table 上，同时避免不必要的全局重渲染。这正是 Valtio 的用武之地。

状态管理层 (Valtio)

Valtio 的 API 极其简洁。我们创建一个 proxy 对象来存储表格数据。对这个对象的任何修改都会被 Valtio 自动追踪，并只触发使用了该部分数据的组件进行重渲染。

// src/store.js
import { proxy } from 'valtio';

// state 的结构设计很重要。使用一个 map (对象字面量) 而不是数组，
// 可以通过主键 O(1) 的时间复杂度快速定位和更新记录。
// 这对于处理大量更新事件至关重要。
const store = proxy({
  orders: {}, // { [orderId]: orderData }
  isConnected: false,
  error: null,
});

// 我们还需要一个派生状态，将 map 转换为 Ant Design Table 需要的数组格式。
// 这个转换可以放在组件内部用 useMemo 实现，或者直接在 store 中管理。
// 这里我们选择在组件中处理，以保持 store 的纯粹性。

export default store;

SSE 客户端服务

封装 EventSource 的逻辑，使其能够与我们的 Valtio store 交互。

// src/services/sseClient.js
import store from '../store';

let eventSource;

export function connectToSSE() {
  if (eventSource) {
    eventSource.close();
  }

  // 这里的 URL 指向我们之前搭建的 Node.js 服务
  eventSource = new EventSource('http://localhost:3001/events');

  eventSource.onopen = () => {
    console.log('SSE connection established.');
    store.isConnected = true;
    store.error = null;
  };

  eventSource.onerror = (err) => {
    console.error('EventSource failed:', err);
    store.isConnected = false;
    store.error = 'Connection lost. Reconnecting...';
    // EventSource 会自动重连，我们只需更新UI状态
  };

  eventSource.addEventListener('cdc_update', (event) => {
    const changes = JSON.parse(event.data);
    
    // 批量处理变更，减少 Valtio 的通知次数
    changes.forEach(change => {
      const { operation, OrderID, ...rowData } = change;
      switch (operation) {
        case 1: // Delete
          if (store.orders[OrderID]) {
            delete store.orders[OrderID];
          }
          break;
        case 2: // Insert
          store.orders[OrderID] = { OrderID, ...rowData };
          break;
        case 4: // Update (new value)
          // 这里的实现很关键：直接修改代理对象的属性，
          // Valtio 会精确地知道哪个对象的哪个字段变了。
          if (store.orders[OrderID]) {
            Object.assign(store.orders[OrderID], { OrderID, ...rowData });
          } else {
            // 如果更新的记录在本地不存在（可能由于初始加载和事件流之间的时间差），
            // 则将其视为插入。
            store.orders[OrderID] = { OrderID, ...rowData };
          }
          break;
        case 3: // Update (old value)
          // 本次场景我们忽略旧值
          break;
        default:
          console.warn('Unknown CDC operation:', operation);
      }
    });
  });
}

// 初始数据加载
export async function loadInitialData() {
    // 在真实应用中，这里会有一个 REST API 用于获取全量数据
    // GET /api/orders
    // fetch('...').then(res => res.json()).then(data => {
    //   store.orders = data.reduce((acc, order) => {
    //     acc[order.OrderID] = order;
    //     return acc;
    //   }, {});
    // });
    // 为演示，我们先初始化为空
    console.log("Initial data loaded (mock). Subscribing to SSE for updates.");
}

React 组件 (Ant Design + Valtio)

// src/components/RealTimeOrderTable.jsx
import React, { useEffect, useMemo } from 'react';
import { Table, Tag, Typography } from 'antd';
import { useSnapshot } from 'valtio';
import store from '../store';
import { connectToSSE, loadInitialData } from '../services/sseClient';

const { Text } = Typography;

const columns = [
  { title: 'Order ID', dataIndex: 'OrderID', key: 'OrderID', sorter: (a, b) => a.OrderID - b.OrderID },
  { title: 'Customer ID', dataIndex: 'CustomerID', key: 'CustomerID' },
  { title: 'Order Date', dataIndex: 'OrderDate', key: 'OrderDate', render: (text) => new Date(text).toLocaleString() },
  { title: 'Total Amount', dataIndex: 'TotalAmount', key: 'TotalAmount', render: (amount) => `$${parseFloat(amount).toFixed(2)}` },
  {
    title: 'Status',
    dataIndex: 'Status',
    key: 'Status',
    render: (status) => {
      let color;
      if (status === 'Shipped') color = 'green';
      else if (status === 'Pending') color = 'gold';
      else if (status === 'Cancelled') color = 'red';
      else color = 'blue';
      return <Tag color={color}>{status.toUpperCase()}</Tag>;
    },
  },
];

const RealTimeOrderTable = () => {
  // useSnapshot 创建了 store 的一个不可变快照。
  // 当 store 的任何部分发生变化，React 会重新渲染此组件。
  // 但 Valtio 的魔力在于，如果子组件只依赖 store 的一小部分，
  // 只有那部分变化时，子组件才会重渲染。
  const snap = useSnapshot(store);

  useEffect(() => {
    // 模拟初始数据加载，然后连接到 SSE
    loadInitialData();
    connectToSSE();
  }, []);

  // 使用 useMemo 来避免在每次渲染时都重新计算 orders 数组。
  // 只有当 snap.orders (即代理对象本身) 发生结构性变化时，
  // (例如添加或删除属性)，才会重新计算。
  // Valtio 保证了这一点。
  const dataSource = useMemo(() => {
    console.log("Recalculating dataSource array from store map...");
    return Object.values(snap.orders).sort((a, b) => b.OrderID - a.OrderID);
  }, [snap.orders]);

  return (
    <div>
      <div style={{ marginBottom: 16, display: 'flex', justifyContent: 'space-between', alignItems: 'center' }}>
        <Typography.Title level={4}>Real-time Order Monitoring</Typography.Title>
        <Tag color={snap.isConnected ? 'success' : 'error'}>
          {snap.isConnected ? 'CONNECTED' : 'DISCONNECTED'}
        </Tag>
      </div>
      {snap.error && <Text type="danger">{snap.error}</Text>}
      <Table
        rowKey="OrderID"
        columns={columns}
        dataSource={dataSource}
        pagination={{ pageSize: 15 }}
        bordered
      />
    </div>
  );
};

export default RealTimeOrderTable;

单元测试的思路：对于后端，可以模拟 mssql 模块，测试 cdcPoller 是否能正确处理不同 LSN 和变更数据，并验证 broadcast 是否被正确调用。对于前端，可以模拟 EventSource，派发各种 cdc_update 事件，然后断言 Valtio store 的状态是否如预期般变化。

整体架构与数据流

整个系统的生命周期可以由下面的图表清晰地展示出来：

sequenceDiagram
    participant User as 用户
    participant DB as SQL Server DB
    participant CDC as CDC Capture Job
    participant Node as Node.js SSE Service
    participant React as React Frontend (Valtio)
    participant Antd as Ant Design Table

    User->>DB: DML 操作 (INSERT, UPDATE, DELETE dbo.Orders)
    DB-->>CDC: 事务日志被捕获
    CDC->>DB: 将变更写入 cdc.dbo_Orders_CT 表

    loop CDC Polling
        Node->>DB: 查询自 last_lsn 以来的变更
        DB-->>Node: 返回变更记录
        Node->>Node: 更新 last_lsn
        Node-->>React: broadcast(changes) via SSE
    end

    React->>React: EventSource 接收到 cdc_update 事件
    React->>React: 解析数据并更新 Valtio store
    Note right of React: Valtio 自动追踪状态变更
    React-->>Antd: 触发最小化重渲染
    Antd-->>User: UI 表格实时更新

局限性与未来优化路径

这个架构虽然高效，但并非没有缺点。

轮询延迟: 后端对 CDC 表的轮询机制引入了一个固有的延迟（本例中为2秒）。虽然远优于前端轮询，但并非真正的零延迟。在 SQL Server 2008 R2 及以上版本中，可以研究使用 SQL Service Broker 和查询通知（Query Notifications）来构建一个推模型，当变更表有新数据时主动通知后端服务，彻底消除轮询。
LSN 持久化: 示例中的 lastLsn 存储在内存中，服务重启会导致其丢失，可能造成数据重复或丢失。在生产环境中，必须将其持久化到 Redis、数据库或一个简单的文件中，并在服务启动时加载。
水平扩展: 当前的 SSE 服务是单点的。如果要水平扩展 Node.js 服务实例，简单的内存 clients 数组将失效。需要引入一个消息总线（如 Redis Pub/Sub），当一个实例从数据库拉取到变更后，通过总线广播给所有其他实例，再由各个实例推送给它们各自连接的客户端。
初始加载与流的同步: 当前端加载初始数据和订阅 SSE 流之间存在一个时间窗口，可能导致数据不一致。一个健壮的方案是：前端请求初始数据时，后端返回当前数据集以及此刻的 LSN。前端加载完数据后，再带上这个 LSN 去订阅 SSE 流，后端则从该 LSN 之后开始推送变更，确保无缝衔接。

SQL Server Ant Design Server-Sent Events (SSE) Valtio

基于 Go 与 Recoil 实现强化学习智能体的实时 UI 交互式训练架构

2023-10-27 分布式架构

强化学习框架 (Go) Recoil

构建基于Scala与Fluentd的弹性数据管道以解决AWS数仓写入瓶颈

2023-10-27 数据工程

Scala AWS Fluentd Data Warehouse