TypeScript中如何使用Observable替代Subject实现事件消息传递？

• 封装性破损：Subject 同时实现了 Observable 和 Observer 接口，对外直接暴露 done 意味着任意外部代码都可以主动调用 done.next()、done.error() 篡改事件流，完全不受 Downloader 类的逻辑控制，会导致不可预期的事件触发。
• 事件丢失风险：当前逻辑和 Promise 强耦合，如果调用 download() 下载完成后，外部才订阅 done，订阅者会完全收不到本次下载完成的通知，不符合事件订阅的预期。
• 逻辑耦合度高：多次调用 download() 时，所有订阅者都会收到所有下载任务的完成通知，无法区分通知对应哪次下载任务，也无法针对单次下载做独立的订阅、取消逻辑。
• 内存泄漏隐患：Subject 没有默认的完成逻辑，长期运行下如果订阅者不主动取消订阅，会持续占用内存。

核心优化思路是：私有化 Subject，只对外暴露只读的 Observable 接口，或直接把异步操作封装为 Observable 返回，符合 RxJS "只在必要场景使用 Subject、对外统一暴露 Observable"的设计规范。

优化后代码示例：

import { Subject, Observable } from 'rxjs';
import { share } from 'rxjs/operators';

class Downloader {
    // 将Subject设为私有属性，禁止外部直接访问修改
    private readonly done$ = new Subject<void>();
    // 对外只暴露只读的Observable，外部只能订阅不能修改事件流
    public readonly done = this.done$.asObservable();

    download(): Observable<void> {
        // 将下载异步逻辑包装为Observable返回
        const downloadTask$ = new Observable<void>(subscriber => {
            downloadSomeFile(...)
                .then(() => {
                    // 触发单次下载的完成通知
                    subscriber.next();
                    subscriber.complete();
                    // 触发全局下载完成事件
                    this.done$.next();
                })
                .catch(err => {
                    subscriber.error(err);
                });
        }).pipe(
            // 多订阅者共享同一个下载请求，不会重复触发下载
            share()
        );

        return downloadTask$;
    }
}

使用方式：

1. 订阅单次下载的完成/错误事件

const downloader = new Downloader();
downloader.download().subscribe({
  next: () => { /* 本次下载完成的自定义逻辑 */ },
  error: (err) => { /* 本次下载的错误处理逻辑 */ }
});

2. 订阅全局所有下载的完成事件

downloader.done.subscribe(() => { /* 任意下载任务完成的全局逻辑 */ });

优化效果：

• 外部获取的所有可订阅对象都是只读的 Observable，无法主动触发事件，事件流完全由 Downloader 内部控制，符合封装原则
• 可灵活扩展能力：如果需要避免后订阅丢失事件，只需添加 shareReplay(1) 操作符即可缓存最近一次完成事件，无需修改核心逻辑
• 单次下载的 Observable 完成/报错后会自动销毁订阅，不会造成内存泄漏
• 可以直接复用 RxJS 生态的所有操作符实现重试、超时、限流等扩展能力，适配不同业务需求

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