面试题：利用 TypeScript 实现高效的大数据流处理

以下是使用 RxJS 在 TypeScript 中处理大量数据流的示例代码：

首先安装 RxJS：

npm install rxjs

示例代码：

import { from, Observable } from 'rxjs';
import { filter, map, reduce } from 'rxjs/operators';

// 模拟每秒产生大量数据的数据流
const generateDataStream = (): Observable<number> => {
    return from(new Array(1000).fill(0).map((_, i) => i + 1));
};

// 处理数据流
const dataStream$ = generateDataStream();

// 过滤出偶数
const filteredData$ = dataStream$.pipe(
    filter((num) => num % 2 === 0)
);

// 将数据乘以 2
const transformedData$ = filteredData$.pipe(
    map((num) => num * 2)
);

// 聚合数据，计算总和
const aggregatedData$ = transformedData$.pipe(
    reduce((acc, num) => acc + num, 0)
);

// 订阅最终结果
aggregatedData$.subscribe((sum) => {
    console.log('聚合结果:', sum);
});

内存管理和性能优化：

避免内存泄漏：在不需要订阅时，及时调用 unsubscribe() 方法取消订阅。例如：

const subscription = aggregatedData$.subscribe((sum) => {
    console.log('聚合结果:', sum);
});

// 当不再需要时
subscription.unsubscribe();

使用 share() 或 shareReplay()：如果多个订阅者需要相同的数据，使用 share() 操作符可以避免重复计算。shareReplay() 还可以缓存一定数量的数据，适用于需要重放数据给新订阅者的场景。

const sharedData$ = dataStream$.pipe(
    filter((num) => num % 2 === 0),
    share()
);

背压处理：如果数据流产生数据的速度比处理速度快，可以使用 bufferTime、windowTime 等操作符来控制数据流入的节奏。例如：

const bufferedData$ = dataStream$.pipe(
    bufferTime(1000), // 每 1 秒缓冲一次数据
    map((buffer) => buffer.reduce((acc, num) => acc + num, 0))
);

这样可以避免因数据堆积导致的性能问题和内存溢出。

星途面试题库

面试题：利用 TypeScript 实现高效的大数据流处理

知识考点

面试题答案