java - 在 RxJava 中取消订阅线程安全吗？

Question

假设我有以下 RxJava 代码(访问数据库，但具体用例无关紧要):

public Observable<List<DbPlaceDto>> getPlaceByStringId(final List<String> stringIds) {
    return Observable.create(new Observable.OnSubscribe<List<DbPlaceDto>>() {
        @Override
        public void call(Subscriber<? super List<DbPlaceDto>> subscriber) {
            try {
                Cursor c = getPlacseDb(stringIds);

                List<DbPlaceDto> dbPlaceDtoList = new ArrayList<>();
                while (c.moveToNext()) {
                    dbPlaceDtoList.add(getDbPlaceDto(c));
                }
                c.close();

                if (!subscriber.isUnsubscribed()) {
                    subscriber.onNext(dbPlaceDtoList);
                    subscriber.onCompleted();
                }
            } catch (Exception e) {
                if (!subscriber.isUnsubscribed()) {
                    subscriber.onError(e);
                }
            }
        }
    });
}

鉴于此代码，我有以下问题:

1. 如果有人取消订阅从此方法返回的可观察对象(在先前的订阅之后)，该操作是否线程安全？那么我的“isUnsubscribed()”检查在这个意义上是否正确，无论日程安排如何？

2. 有没有比我在这里使用的方法更简洁、使用更少样板代码来检查未订阅状态的方法？我在框架中找不到任何东西。我以为 SafeSubscriber 解决了订阅者取消订阅时不转发事件的问题，但显然它没有。

Answer 1

is that operation thread-safe?

是的。您收到一个 rx.Subscriber ，它(最终)检查一个 volatile boolean 值，当订阅者的订阅被取消订阅时，该 boolean 值被设置为 true 。

cleaner way with less boilerplate code to check for unsubscribed states

SyncOnSubscribe 和 AsyncOnSubscribe(自 1.0.15 版起作为 @Experimental api 提供)是为此用例创建的。它们可以作为调用 Observable.create 的安全替代方法。这是同步情况的(人为的)示例。

public static class FooState {
    public Integer next() {
        return 1;
    }
    public void shutdown() {

    }
    public FooState nextState() {
        return new FooState();
    }
}
public static void main(String[] args) {
    OnSubscribe<Integer> sos = SyncOnSubscribe.createStateful(FooState::new, 
            (state, o) -> {
                o.onNext(state.next());
                return state.nextState();
            }, 
            state -> state.shutdown() );
    Observable<Integer> obs = Observable.create(sos); 
}

请注意，SyncOnSubscribe 下一个函数不允许在每次迭代中多次调用 observer.onNext，也不能同时调用该观察者。这里有几个指向 SyncOnSubscribe implementation 的链接和 tests在 1.x 分支的头部。它的主要用途是简化编写同步迭代或解析数据的 observable 和 onNext 下游，但在支持背压并检查是否取消订阅的框架中这样做。本质上，您将创建一个 next 函数，每当下游运算符(operator)需要一个新的数据元素 onNexted 时，该函数就会被调用。您的下一个函数可以调用 onNext 0 次或 1 次。

AsyncOnSubscribe 旨在很好地处理异步操作的可观察源(例如开箱调用)的背压。您的下一个函数的参数包括请求计数，并且您提供的可观察对象应该提供一个可满足该请求数量的数据的可观察对象。这种行为的一个例子是来自外部数据源的分页查询。

以前，将 OnSubscribe 转换为 Iterable 并使用 Observable.from(Iterable) 是一种安全的做法。此实现获取迭代器并为您检查 subscriber.isUnsubscribed()。