为啥 onSubscribe 在 rxjava 中不起作用？

Posted 2023-03-22

【中文标题】为啥 onSubscribe 在 rxjava 中不起作用？

【英文标题】：Why onSubscribe does not work in rxjava?为什么 onSubscribe 在 rxjava 中不起作用？

【发布时间】：2017-06-20 13:05:46

【问题描述】：

如果我不写observeOn 行，当我在代码下面运行时，应用程序会崩溃，因为getView().showBlockLayout(isBlock); 调用了一个尝试隐藏或显示布局的方法。 但我尝试将observeOn(androidSchedulers.mainThread()) 下方更改为subscribeOn(AndroidSchedulers.mainThread()) 并且应用再次崩溃！

subscription.add(UserStore.getInstance().getBlockObservable(databaseHelper.getConference().getUserChatId())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<Boolean>() 
                @Override
                public void call(Boolean isBlock) 
                    getView().showBlockLayout(isBlock);
                    databaseHelper.getConference().setBlock(isBlock);
                    mConferenceModel.setBlock(isBlock);
                
            ));

我也对此进行了测试：

subscription.add(UserStore.getInstance().getBlockObservable(databaseHelper.getConference().getUserChatId())
            .subscribeOn(Schedulers.computation())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<Boolean>() 
                @Override
                public void call(Boolean isBlock) 
                    getView().showBlockLayout(isBlock);
                    databaseHelper.getConference().setBlock(isBlock);
                    mConferenceModel.setBlock(isBlock);
                
            ));

出乎意料的是它起作用了并且没有崩溃！我没有在 getBlockObservable 方法中使用 subscribeOn（因为我知道我们可以设置一次）

这是我的UserStore 课程

PublishSubject<Pair<String,Boolean>> mObservableBlock;

private UserStore()
    mObservableBlock = PublishSubject.create();
    mInstance = this;


public static UserStore getInstance() 
    if(mInstance == null)
        new UserStore();
    return mInstance;


public Observable<Boolean> getBlockObservable(final String userId)
    return mObservableBlock
            .observeOn(Schedulers.computation())
            .filter(new Func1<Pair<String,Boolean>, Boolean>() 
        @Override
        public Boolean call(Pair<String,Boolean> s) 
            if(userId.equals(s.first))
                return true;
            return false;
        
    ).map(new Func1< Pair<String, Boolean>, Boolean>() 

        @Override
        public Boolean call(Pair<String, Boolean> UserBlock) 
            return UserBlock.second;
        
    );

public void publishBlockedUser(String userId,boolean isBlock)
    mObservableBlock.onNext(new Pair<String, Boolean>(userId,isBlock));

这是我在 gradle 中导入 rxjava 依赖项的方式

compile 'io.reactivex:rxjava:1.1.5'
compile 'io.reactivex:rxandroid:1.2.0'

【问题讨论】：

当您将.observeOn(AndroidSchedulers.mainThread()) 更改为subscribeOn（它仍然崩溃）时，您传递给subscribeOn 调用的参数是什么？

@theFunkyEngineer subscribeOn(AndroidSchedulers.mainThread()) 如果我没有 observeOn 它会崩溃

它崩溃了，因为我尝试在 UI 线程中更改某些内容

【参考方案1】：

正如medium artice中提到的：

一个重要的事实是 subscribeOn 不适用于主题。

所以你不能对主题使用 subscribeOn，我们必须在订阅前使用observerOn(AndroidSchedulers.mainThread())。 所以之后所有的下游方法都会在 mainThread 上调用。

检查这个medium artice