跳到主要内容

异步数据处理

DarraRT 的 HMI 常面临 高频变量 场景: PLC 每秒推送数百个变量、趋势图累积数万点、报警列表滚动不停。本章讲如何在浏览器前端异步处理这些数据流, 避免 UI 卡顿。

注: 这里的"数据处理"是 PLC 变量的前端处理, 不是业务数据 ETL, 也不涉及订单/客户/物料等业务管道。

卡顿根源​

典型卡顿来源:

根源症状解法
主线程 JS 计算阻塞点击无响应, 动画卡Web Worker
DOM 频繁更新滚动抖动, FPS 掉requestAnimationFrame / 虚拟列表
Canvas 重绘趋势图更新慢双缓冲 / OffscreenCanvas
内存泄漏数小时后页面崩溃IndexedDB 归档 + 定期 GC
网络反复重连数据跳跃darra.on('offline') 切到缓存

darra.bus: 内置事件总线​

darra-plc.js 在 v1.5+ 暴露了 darra.bus 作为内部事件总线, 多组件共享同一份订阅, 避免重复绑定:

// 模块 A
darra.bus.on('var:MD.Temp', v => updateGauge(v));

// 模块 B
darra.bus.on('var:MD.Temp', v => updateTrend(v));

// 模块 C 只在需要时订阅, 不需要时 off
const handler = v => updateExport(v);
darra.bus.on('var:MD.Temp', handler);
// ... 关闭面板时
darra.bus.off('var:MD.Temp', handler);

框架内部把 N 个 bind 合并成一次 WebSocket 订阅, 降低服务端开销。

Web Worker 处理趋势数据​

10 万点趋势直接在主线程 downsample 会冻结 UI 数百毫秒。迁移到 Worker:

Worker 侧 (trend-worker.js)​

// 接收主线程消息
self.onmessage = (e) => {
    const { cmd, data } = e.data;

    switch (cmd) {
        case 'downsample':
            // 输入 100000 点 [{t, v}, ...], 输出 1000 点
            const result = lttb(data.points, 1000);  // Largest-Triangle-Three-Buckets
            self.postMessage({ cmd: 'downsampled', result });
            break;

        case 'stats':
            const stats = computeStats(data.points);
            self.postMessage({ cmd: 'stats', stats });
            break;
    }
};

function lttb(points, targetCount) {
    if (points.length <= targetCount) return points;
    // LTTB 算法: 每个桶保留最能代表形状的点
    // ... 省略实现 ...
    return sampled;
}

function computeStats(points) {
    let min = Infinity, max = -Infinity, sum = 0;
    for (const p of points) {
        if (p.v < min) min = p.v;
        if (p.v > max) max = p.v;
        sum += p.v;
    }
    return { min, max, avg: sum / points.length, count: points.length };
}

主线程调用​

const worker = new Worker('/static/js/trend-worker.js');

let pending = null;
worker.onmessage = (e) => {
    const { cmd, result } = e.data;
    if (cmd === 'downsampled') {
        drawTrend(result);    // 1000 点, Canvas 很快
        pending = null;
    }
};

function onPointsArrive(newPoints) {
    allPoints.push(...newPoints);
    if (allPoints.length > 200_000) {
        // 超 20 万点截断前半
        allPoints = allPoints.slice(100_000);
    }
    if (!pending) {
        pending = true;
        worker.postMessage({ cmd: 'downsample', data: { points: allPoints }});
    }
}

Worker 处理期间 UI 完全不冻结, 动画仍 60 FPS.

IndexedDB 缓存历史​

离线时让用户仍能看到"最后 N 分钟"数据:

// 打开数据库
const db = await new Promise((res, rej) => {
    const req = indexedDB.open('darra-cache', 1);
    req.onupgradeneeded = (e) => {
        const d = e.target.result;
        if (!d.objectStoreNames.contains('points')) {
            const os = d.createObjectStore('points', { keyPath: 'id', autoIncrement: true });
            os.createIndex('by_var_ts', ['var', 'ts']);
        }
    };
    req.onsuccess = e => res(e.target.result);
    req.onerror   = e => rej(e.target.error);
});

// 保存一批点
async function saveBatch(records) {
    return new Promise((res, rej) => {
        const tx = db.transaction('points', 'readwrite');
        const os = tx.objectStore('points');
        for (const r of records) os.add(r);
        tx.oncomplete = res;
        tx.onerror    = e => rej(e.target.error);
    });
}

// 查询过去 N 秒
async function queryRecent(varName, seconds) {
    const since = Date.now() - seconds * 1000;
    return new Promise((res, rej) => {
        const tx = db.transaction('points');
        const idx = tx.objectStore('points').index('by_var_ts');
        const range = IDBKeyRange.bound([varName, since], [varName, Infinity]);
        const out = [];
        idx.openCursor(range).onsuccess = (e) => {
            const cur = e.target.result;
            if (cur) { out.push(cur.value); cur.continue(); }
            else res(out);
        };
        tx.onerror = e => rej(e.target.error);
    });
}

结合 WebSocket​

let buffer = [];
darra.on('variable', ({ name, value, ts }) => {
    buffer.push({ var: name, v: value, ts: ts || Date.now() });
    if (buffer.length >= 500) {
        saveBatch(buffer);
        buffer = [];
    }
});

darra.on('offline', async () => {
    const lastTemp = await queryRecent('MD.Temp', 300);   // 过去 5 分钟
    drawTrend(lastTemp);
    showBanner('离线模式, 显示最后 5 分钟缓存');
});

去抖与节流​

高频变量 HMI 不需要每次变化都重绘:

去抖 (debounce)​

function debounce(fn, ms) {
    let t = null;
    return (...args) => {
        clearTimeout(t);
        t = setTimeout(() => fn(...args), ms);
    };
}

const saveConfig = debounce(() => {
    darra.write('DB_Config.Value', currentValue);
}, 500);

// 用户拖滑块, 停止 500ms 后才写入 PLC
slider.addEventListener('input', e => {
    currentValue = e.target.value;
    saveConfig();
});

节流 (throttle)​

function throttle(fn, ms) {
    let last = 0, pending = null;
    return (...args) => {
        const now = Date.now();
        if (now - last >= ms) {
            last = now; fn(...args);
        } else if (!pending) {
            pending = setTimeout(() => {
                last = Date.now(); pending = null; fn(...args);
            }, ms - (now - last));
        }
    };
}

const updateGauge = throttle(value => gaugeEl.update(value), 100);
darra.bind('MD.Temp', updateGauge);

虚拟列表​

报警列表 10000 条同时渲染会卡死 DOM。虚拟列表只渲染可见的 20 行:

<div class="virtual-list" x-data="virtualList(alarms, 32)"
     @scroll="onScroll"
     :style="`height: 640px; overflow-y: auto`">

    <div :style="`height: ${totalHeight}px; position: relative`">
        <template x-for="(a, i) in visibleItems" :key="a.id">
            <div class="row"
                 :style="`position: absolute; top: ${(startIdx + i) * 32}px; height: 32px`">
                <span x-text="a.time"></span>
                <span x-text="a.message"></span>
            </div>
        </template>
    </div>
</div>
function virtualList(items, rowHeight) {
    return {
        items,
        rowHeight,
        scrollTop: 0,
        viewHeight: 640,
        get totalHeight()  { return this.items.length * this.rowHeight; },
        get startIdx()     { return Math.floor(this.scrollTop / this.rowHeight); },
        get visibleItems() {
            const count = Math.ceil(this.viewHeight / this.rowHeight) + 2;
            return this.items.slice(this.startIdx, this.startIdx + count);
        },
        onScroll(e)        { this.scrollTop = e.target.scrollTop; }
    };
}

10000 条 <= 100 行 DOM 活跃, 滚动丝滑。

requestAnimationFrame​

所有重绘聚集到下一帧, 避免一次性多次 reflow:

let queued = false;
let pendingData = null;

darra.bind('MD.Temp', v => {
    pendingData = v;
    if (!queued) {
        queued = true;
        requestAnimationFrame(() => {
            gaugeEl.textContent = pendingData.toFixed(1);
            queued = false;
        });
    }
});

浏览器 16.7 ms 一帧, 期间多次变更只合并成一次 DOM 更新。

OffscreenCanvas​

Chrome 支持把 Canvas 渲染也搬到 Worker, 主线程不掉帧:

// 主线程
const canvas = document.getElementById('trend');
const off = canvas.transferControlToOffscreen();
worker.postMessage({ cmd: 'init', canvas: off }, [off]);

// Worker 内
let ctx = null;
self.onmessage = (e) => {
    if (e.data.cmd === 'init') ctx = e.data.canvas.getContext('2d');
    else if (e.data.cmd === 'points') drawPoints(ctx, e.data.points);
};

内存监控​

长时间运行 HMI 要防内存泄漏:

if (performance.memory) {
    setInterval(() => {
        const mb = performance.memory.usedJSHeapSize / 1024 / 1024;
        if (mb > 500) {
            console.warn('内存占用', mb.toFixed(0), 'MB');
            // 触发清理: 清空旧趋势数据、关闭不活动面板
            purgeOldPoints();
        }
    }, 30_000);
}

性能测试基线​

场景目标测量方式
订阅 500 个变量CPU <20%Chrome DevTools Performance
10 万点趋势首绘 <200ms, 更新 <50msLTTB in Worker
报警列表 10000 条滚动 60 FPS虚拟列表
页面连续运行 24h内存增长 <100MBperformance.memory
切换页面<500msChrome Performance

相关文档​