在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. 如果有人从此方法返回的observable取消订阅(在之前的订阅之后),该操作是否是线程安全的？那么无论调度如何,我的'isUnsubscribed()'检查在这个意义上是否正确？

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

Answer 1

那个操作线程安全吗？

是.您正在接收一个rx.Subscriber,它(最终)检查订阅者订阅取消订阅时设置为true的volatile布尔值.

更简洁的方法,用更少的样板代码来检查未订阅的状态

在SyncOnSubscribe与AsyncOnSubscribe(作为@ExperimentalAPI的版本1.0.15的)被用于该用途的情况下创建的.它们可以作为呼叫的安全替代方案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 实施和测试的几个链接1.x.它的主要用途是简化编写同步和onNext下游迭代或解析数据的observable,但在支持背压的框架中执行此操作,并检查是否取消订阅.基本上,您将创建一个next函数,每当下游运算符需要一个新的数据元素onNexted时,就会调用该函数.你的下一个函数可以调用onNext 0或1次.

它AsyncOnSubscribe的设计可以很好地处理背压,以便异步操作的可观察源(例如外线呼叫).您的下一个函数的参数包括请求计数,您提供的observable应该提供一个observable,它可以满足所请求数量的数据.此行为的一个示例是来自外部数据源的分页查询.

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