RxJS技术问询：无需Observable完成使用forkJoin及复用已完成Observable

问题解答 #

1. 能否在底层Observable未完成时触发forkJoin()？ #

不行。forkJoin的核心设计逻辑就是等待所有传入的Observable都完成，再收集每个Observable的最后一个值。只要有一个源Observable未完成，forkJoin就不会发射任何值，这是它的固有特性，无法绕过。

2. 能否复用已完成的Observable？ #

可以，但针对你需要重复点击发起请求的场景，更合理的方式是重新设计流的结构，避免依赖已完成的Observable。通过将forkJoin嵌套在点击事件流内部，每次点击都重新执行一次forkJoin，既满足“等待所有请求完成再更新UI”的需求，又支持重复触发。

修改后的代码 #

JavaScript 部分 #

import { catchError, exhaustMap, forkJoin, fromEvent, of, tap } from 'rxjs';
import { ajax } from "rxjs/ajax";
import { map } from "rxjs/operators";

const observer = {
    next: val => addImages({ dog: val.dog, cat: val.cat }),
    error: err => console.log('error handled in observer'),
    complete: () => console.log('complete!')
};

const fetchButton = document.querySelector('.fetchImages');

// 点击事件触发，每次点击都发起一组新的请求组合
fromEvent(fetchButton, 'click').pipe(
    exhaustMap(() => 
        forkJoin({
            dog: ajax.getJSON('http://localhost:3000/dog').pipe(
                map(item => item.message),
                catchError(error => {
                    console.log('error, handling DOG error locally');
                    return of('https://world.openfoodfacts.org/images/products/899/110/278/9318/front_en.9.full.jpg');
                })
            ),
            cat: ajax.getJSON('https://some-random-api.ml/img/cat').pipe(
                map(item => item.link),
                catchError(error => {
                    console.log('error, handling CAT error locally');
                    return of('https://world.openfoodfacts.org/images/products/930/065/703/2029/front_en.3.full.jpg');
                })
            )
        })
    ),
    tap(console.log)
).subscribe(observer);

function addImages({ dog, cat }) {
    document.querySelector('[alt="dog"]').setAttribute('src', dog);
    document.querySelector('[alt="cat"]').setAttribute('src', cat);
}

HTML 部分（无需改动） #

<section>
    <button class="fetchImages">Fetch!</button>

    <div class="images">
        <img alt="dog">
        <img alt="cat">
    </div>
</section>

关键改动说明 #

• 移除了独立的dogSource$、catSource$等流，直接将点击事件作为触发源，每次点击通过exhaustMap发起新的forkJoin请求组合，避免了原代码中take(1)导致流完成后无法重复触发的问题。
• exhaustMap会忽略当前请求未完成时的重复点击，防止重复发起请求，保留了你原代码中的需求逻辑。
• 错误处理和数据映射逻辑内嵌在forkJoin的源Observable中，结构更清晰，且每次点击都会重新执行，不会因流完成而失效。

如果确实需要复用已完成Observable的结果（比如缓存请求数据），可以使用shareReplay(1)操作符缓存最后一个值，示例如下：

const cachedDogImage$ = ajax.getJSON('http://localhost:3000/dog').pipe(
    map(item => item.message),
    catchError(error => {
        console.log('error, handling DOG error locally');
        return of('https://world.openfoodfacts.org/images/products/899/110/278/9318/front_en.9.full.jpg');
    }),
    shareReplay(1)
);

但该方案更适合缓存场景，对于你每次点击需要新数据的需求，嵌套forkJoin的方案更贴合。

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