RxJS技术咨询：是否存在可拼接映射后元素的流操作符？

首先直接给结论：RxJS里确实没有现成的单一操作符能直接完成你想要的「先输出原流所有元素，再输出每个元素映射后的结果」这个行为，但我们可以通过组合现有操作符或者封装自定义操作符来实现，而且能比你当前的写法更简洁，避免拆分存储流的麻烦。

为什么现有操作符不适用？ #

• concatMap会把原流的每个元素替换成映射后的结果，相当于「替换」而非「追加」，所以会丢失原元素；
• concat（或者RxJS 7+的concatWith）是合并多个流，但它本身不能直接对单个流的元素做映射后再合并，需要先单独处理映射后的流。

解决方案1：用concatWith简化现有逻辑 #

如果你不想写自定义操作符，RxJS 7及以上版本提供了pipeable的concatWith操作符，可以直接在管道里合并原流和它的映射流：

import { of } from 'rxjs';
import { concatWith, map } from 'rxjs/operators';

// 假设你的f函数是这样的
const f = (x: string) => `f(${x})`;

const item$ = of("A","B","C");
item$.pipe(
  concatWith(item$.pipe(map(x => f(x))))
).subscribe(val => console.log(val));
// 输出顺序：A → B → C → f(A) → f(B) → f(C)

如果你的流是热流（比如来自DOM事件、WebSocket等），直接这样写会导致原流被订阅两次，这时候需要用shareReplay来共享订阅：

import { of, shareReplay } from 'rxjs';
import { concatWith, map } from 'rxjs/operators';

const item$ = of("A","B","C").pipe(shareReplay()); // 共享订阅避免重复触发
item$.pipe(
  concatWith(item$.pipe(map(x => f(x))))
).subscribe(val => console.log(val));

解决方案2：封装自定义操作符（推荐长管道场景） #

如果这个逻辑需要在多个管道里复用，封装成自定义操作符会更优雅，不用重复写合并流的代码：

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

// 自定义操作符：追加每个元素映射后的结果到原流末尾
function appendMapped<T, R>(project: (value: T) => R): (source: Observable<T>) => Observable<T | R> {
  return (source) => concat(source, source.pipe(map(project)));
}

// 使用示例
const f = (x: string) => `f(${x})`;
const item$ = of("A","B","C");

item$.pipe(appendMapped(x => f(x)))
  .subscribe(val => console.log(val));
// 输出顺序和之前一致

这个自定义操作符可以直接插入任何RxJS管道中，完全不用拆分存储原流，非常适合长管道场景。

额外说明 #

如果你想要的是「每个元素先输出原元素，再输出映射后的元素」（比如A→f(A)→B→f(B)→C→f(C)），那可以用concatMap(x => of(x, f(x)))，但这和你描述的需求不一样，这里只是补充提一下避免混淆。

内容的提问来源于stack exchange，提问作者ᴘᴀɴᴀʏɪᴏᴛɪs