在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

操作是线程安全的吗?

是的。你正在接收一个rx。订阅者，它(最终)根据一个易失性布尔值进行检查，该布尔值在订阅者的订阅被取消订阅时设置为true。

使用更少的样板代码来检查未订阅状态

为这个用例创建了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 next函数不允许在每次迭代中调用observer.onNext超过一次，也不能并发地调用该观察者。这里有几个SyncOnSubscribe实现的链接，以及1.x分支头部的测试。它的主要用途是简化编写对数据进行同步迭代或解析的可观察对象和onNext下游，但要在支持背压和检查是否取消订阅的框架中完成。实际上，您将创建一个next函数，每当下游操作符需要onnext的新数据元素时，都会调用该函数。下一个函数可以调用onNext 0或1次。

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

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