RxJava中调用unsubscribe触发线程中断？Subscription相关疑问

1. 两个Subscription实例不同，为何调用外部的unsubscribe能触发线程中断？ #

你看到的外部SafeSubscriber是RxJava自动包装的结果——当你调用subscribe(x->System.out.println(x))时，RxJava会把你传入的Lambda表达式包装成SafeSubscriber（它实现了Subscription接口），作为对外暴露的订阅实例。

而内部的BooleanSubscription，是通过subscriber.add(subscription)这行代码被添加到了SafeSubscriber的内部订阅集合中。当你调用外部SafeSubscriber的unsubscribe()方法时，它会遍历所有添加到自身的Subscription实例，逐个调用它们的unsubscribe()方法。所以内部BooleanSubscription的取消逻辑（也就是线程中断）会被触发，最终实现立即中断线程的效果。

2. 是否存在一个Subscriber对应两个不同Subscription的情况？ #

当然存在。Subscriber接口（包括SafeSubscriber）提供了add(Subscription)方法，允许你向同一个Subscriber实例添加多个不同的Subscription。

比如你可以在Observable的创建逻辑里多次调用subscriber.add()，同时管理多个资源的订阅：

Observable.create(subscriber -> {
    // 第一个线程及对应订阅
    Thread thread1 = new Thread(() -> {
        sleep(10, SECONDS);
        if (!subscriber.isUnsubscribed()) {
            subscriber.onNext(1);
        }
    });
    thread1.start();
    subscriber.add(Subscriptions.create(thread1::interrupt));
    
    // 第二个线程及对应订阅
    Thread thread2 = new Thread(() -> {
        sleep(15, SECONDS);
        if (!subscriber.isUnsubscribed()) {
            subscriber.onNext(2);
        }
    });
    thread2.start();
    subscriber.add(Subscriptions.create(thread2::interrupt));
});

当这个Subscriber被取消订阅时，所有添加进去的Subscription都会被依次执行unsubscribe()，从而释放多个关联资源。

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