RxJS 7中如何在subscribe回调获取完成结果，无需额外Subject

你当前的问题核心是误用了throttle操作符：throttle是限流操作符，仅用于控制源事件的放行时机，它不会把durationSelector返回的异步结果向下游传递，所以你拿不到mockHTTPRequest的返回值，才不得不额外声明了一个Subject来中转结果。

如果你需要的是「按节流规则放行任务，执行异步请求后拿到返回结果」，可以调整操作符组合为throttle + concatMap，或者如果你的需求是串行执行所有任务（前一个请求完成再处理下一个），直接用concatMap即可满足限流需求。

修改后的完整代码 #

import colors from 'colors/safe';
import { Subject, from } from 'rxjs';
import { throttle, concatMap } from 'rxjs/operators';

interface IEventTask {
    id: string,
    createdTime: number
}

let global_counter: number = 0;

// 仅保留纯请求逻辑，移除对外部Subject的依赖
const mockHTTPRequest = async (event: IEventTask) => {
    await new Promise((resolve) => {
        setTimeout(resolve, 1000);
    });

    if (event.id === '01') {
        throw new Error(`error: ${event.id}`);
    }

    return `result: ${event.id}`;
}

const subject = new Subject<IEventTask>();
subject
    .pipe(
        // 原有节流逻辑保留，控制放行时机
        throttle(event => mockHTTPRequest(event), {
            leading: true, trailing: true
        }),
        // 对每个放行的任务，执行请求，将结果向下传递
        concatMap(event => from(mockHTTPRequest(event).catch(err => {
            console.error(err);
            return err.message;
        })))
    )
    .subscribe({
        next: (result) => {
            // 这里直接拿到请求结果
            console.log(`${colors.green(`finished`)} ${result} at: ${global_counter++}`);
        },
        error: (error) => {
            console.error(error);
        },
        complete: () => {
            console.log(`completed`);
        }
    });

// 触发任务的逻辑不变
(async () => {
    for (let i = 1; i <= 10; i++) {
        await new Promise((resolve) => {
            setTimeout(() => {
                resolve(true);
            }, 125);
        });

        const value: IEventTask = {
            id: (i).toString().padStart(2, '0'),
            createdTime: global_counter++
        };

        subject.next(value);
    }
})();

补充说明 #

• 如果你不需要throttle的leading/trailing特殊逻辑，只是要串行执行所有任务，可以直接删除throttle操作符，仅保留concatMap即可，它会自动等前一个请求完成再处理下一个任务，天然起到限流效果。
• 所有请求的结果会直接流入subscribe的next回调，不需要额外的Subject做中转，满足你简化代码的需求。

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