如何使用RxJS从嵌套类型对象数组中获取全局最大最小值？

问题根源 #

你的实现存在三个核心问题：

1. 异步逻辑竞态：RxJS的subscribe是异步执行的，函数在Observable还未处理完所有数据时就直接返回了初始值，最终返回的是最后一次订阅回调更新的行级最值，而非全局最值。
2. 比较函数使用错误：RxJS的max/min操作符要求比较函数返回差值数字（负数/0/正数），而非直接返回比较的数值，这会导致比较逻辑完全失效。
3. 逻辑层级缺失：原代码仅提取了每行的最值并依次覆盖变量，没有对所有行的最值再做全局聚合计算。

修正后的实现 #

以下提供两种可行方案，均解决了异步问题并修正了比较逻辑：

方案1：扁平化所有数据后直接取全局最值（简洁高效） #

import { of, from, mergeMap, max, min } from 'rxjs';
import { lastValueFrom } from 'rxjs';

async function findMax(column: DataTableType): Promise<number> {
  const observable = of(...column).pipe(
    mergeMap(item => from(item.series)), // 扁平化所有series数据项
    max((a, b) => a.value - b.value) // 正确的比较逻辑：返回差值判断大小
  );
  const result = await lastValueFrom(observable);
  return result.value;
}

async function findMin(column: DataTableType): Promise<number> {
  const observable = of(...column).pipe(
    mergeMap(item => from(item.series)),
    min((a, b) => a.value - b.value)
  );
  const result = await lastValueFrom(observable);
  return result.value;
}

方案2：先取行级最值，再聚合全局最值（保留行级处理逻辑） #

import { of, from, mergeMap, max, min, map } from 'rxjs';
import { lastValueFrom } from 'rxjs';

async function findMax(column: DataTableType): Promise<number> {
  const observable = of(...column).pipe(
    mergeMap(item => 
      from(item.series).pipe(
        max((a, b) => a.value - b.value), // 提取当前行的最大值
        map(rowMax => rowMax.value) // 转换为纯数值
      )
    ),
    max((a, b) => a - b) // 对所有行的最大值取全局最大值
  );
  return await lastValueFrom(observable);
}

async function findMin(column: DataTableType): Promise<number> {
  const observable = of(...column).pipe(
    mergeMap(item => 
      from(item.series).pipe(
        min((a, b) => a.value - b.value), // 提取当前行的最小值
        map(rowMin => rowMin.value)
      )
    ),
    min((a, b) => a - b) // 对所有行的最小值取全局最小值
  );
  return await lastValueFrom(observable);
}

使用示例 #

// 异步调用需使用await或.then处理结果
async function test() {
  const globalMax = await findMax(datatable);
  const globalMin = await findMin(datatable);
  console.log('全局最大值:', globalMax); // 输出9120
  console.log('全局最小值:', globalMin); // 输出-9111
}

test();

关键修正说明 #

• 异步处理：使用lastValueFrom将Observable转换为Promise，确保等待所有数据处理完成后再返回结果，彻底解决异步竞态问题。
• 比较逻辑修正：a.value - b.value的返回值符合RxJS要求：
  • 正数表示a.value > b.value，max会保留a；
  • 负数表示a.value < b.value，max会保留b；
  • 0表示两者相等，不影响结果。
• 全局聚合：两种方案均实现了对所有数据的全局最值计算，而非仅处理最后一行数据。

内容的提问来源于stack exchange，提问作者amachree tamunoemi