JavaScript 中怎么在数据流清洗过程中实现流式去重-js教程-PHP中文网

javascript流式去重核心是边接收边判断，用set或lru等轻量结构维护已见集合，结合async iterable或transformstream逐项处理，支持ttl、容量限制及分布式扩展。

javascript 中怎么在数据流清洗过程中实现流式去重

JavaScript 中实现流式去重，核心是边接收数据边判断是否已存在，避免等待全部数据到达再处理，从而降低内存占用、提升实时性。关键在于维护一个轻量、可增量更新的“已见集合”，并配合异步流（如 ReadableStream、RxJS Observable、Node.js Stream 或 async iterable）做逐项处理。

用 Set + async iterable 实现基础流式去重

适用于浏览器或 Node.js 环境中使用 for await...of 消费异步迭代器的场景（如 Fetch 流、自定义生成器）：

用 Set 存储已出现过的唯一键（推荐用字符串键，避免对象直接作 key）
对每条数据提取去重依据（如 item.id、item.email），检查是否在 Set 中
未见过则加入 Set 并 yield/转发；已存在则跳过

示例：

async function* dedupeStream(asyncIterable, keyFn = x => x) {
  const seen = new Set();
  for await (const item of asyncIterable) {
    const key = keyFn(item);
    if (!seen.has(key)) {
      seen.add(key);
      yield item;
    }
  }
}
<p>// 使用：假设有按行读取的文本流（每行是一个 JSON 对象）
const lineStream = /<em> ... </em>/;
for await (const uniqueItem of dedupeStream(lineStream, x => x.userId)) {
console.log(uniqueItem); // 只输出 userId 首次出现的项
}</p>

处理高吞吐或长期运行场景：带过期与容量限制

纯 Set 会无限增长，不适合长时间运行或数据量大的流（如日志、事件流）。此时需引入：

TTL 过期：用 Map + 时间戳 + 定期清理（或惰性检查），或借助 lru-cache 等库
固定容量 LRU 缓存：自动淘汰最久未用的 key，适合热点数据去重
布隆过滤器（Bloom Filter）：内存极省、支持超大数据集，但有极低误判率（可能漏掉极少数重复项，不误删）——适合「允许少量重复」的场景

简单 LRU 示例（使用 lru-cache）：

Java Maven Code Review

审查Java Maven项目（ZIP压缩包或GitLab仓库URL），检查代码规范、命名、模块边界、可维护性问题以及重复代码。

下载

import { LRUCache } from 'lru-cache';
const cache = new LRUCache({ max: 10000 }); // 最多缓存 1 万个 key
<p>async function* lruDedupe(iterable, keyFn) {
for await (const item of iterable) {
const key = keyFn(item);
if (!cache.has(key)) {
cache.set(key, true);
yield item;
}
}
}</p>

与标准流（Node.js / Web Streams）集成

若使用 Node.js ReadableStream 或 Web Streams API，需封装为 TransformStream：

在 transform() 中执行去重逻辑
内部仍用 Set / LRU / Bloom Filter 维护状态
注意背压：确保 controller.enqueue() 不阻塞，且错误要正确传递

简易 TransformStream 版本：

function createDedupeTransform(keyFn = x => x) {
  const seen = new Set();
  return new TransformStream({
    transform(chunk, controller) {
      const key = keyFn(chunk);
      if (!seen.has(key)) {
        seen.add(key);
        controller.enqueue(chunk);
      }
    }
  });
}
<p>// 使用
const response = await fetch('/events');
const body = response.body.pipeThrough(createDedupeTransform(x => x.eventId));
for await (const chunk of body) {
// 已去重的 chunk
}</p>