RxJava：observeOn，subscribeOn和doFinally，在IO和UI线程之间切换

2 回答

• 2

  您需要做的就是将 observeOn 向上移动 . observeOn 方法更改了调用 onNext ， onError 和 onCompleted 的线程，其内部操作和副作用如何工作（通过提升）

  listUseCase.execute(null)
              .subscribeOn(schedulerProvider.io()) // Move subscribe on here
              .observeOn(schedulerProvider.main()) // Change threads here
              .doOnSubscribe(new Consumer<Disposable>() {
                  @Override
                  public void accept(@NonNull Disposable disposable) throws Exception {
                      getView().showLoading(true); // This should be on the main thread also
                  }
              })
              .doFinally(new Action() {
                  @Override
                  public void run() throws Exception {
                      getView().showLoading(false);
                  }
              })
  
              .subscribe(
                      new Consumer<List<AccountEntity>>() {
                          @Override
                          public void accept(@NonNull List<AccountEntity> accountEntities) throws Exception {
                              getView().setAccounts(accountEntities);
                          }
                      },
                      new Consumer<Throwable>() {
                          @Override
                          public void accept(@NonNull Throwable throwable) throws Exception {
                              if (isViewAttached()) {
                                  getView().showError(throwable.getMessage());
                              }
                          }
                      }
              );
  

  回复于 2024-05-03T17:36:17+08:00
• 2

  问题出现了，因为我没有在活动完成/销毁时处理订阅 .

  现在，每个活动/视图告诉演示者何时停止或销毁它们并且演示者处理订阅 .

  这似乎解决了我的问题 .

  @Override
  public void initialize() {
      if (!isViewAttached()) {
          throw new ViewNotAttachedException();
      }
      disposable = listUseCase.execute(null)
              .subscribeOn(schedulerProvider.io()) // Move subscribe on here
              .observeOn(schedulerProvider.main()) // Change threads here
              .doOnSubscribe(new Consumer<Disposable>() {
                  @Override
                  public void accept(@NonNull Disposable disposable) throws Exception {
                      getView().showLoading(true); // This should be on the main thread also
                  }
              })
              .doFinally(new Action() {
                  @Override
                  public void run() throws Exception {
                      getView().showLoading(false);
                  }
              })
              .subscribe(
                      new Consumer<List<AccountEntity>>() {
                          @Override
                          public void accept(@NonNull List<AccountEntity> accountEntities) throws Exception {
                              getView().setAccounts(accountEntities);
                          }
                      },
                      new Consumer<Throwable>() {
                          @Override
                          public void accept(@NonNull Throwable throwable) throws Exception {
                              if (isViewAttached()) {
                                  getView().showError(throwable.getMessage());
                              }
                          }
                      }
              );
  }
  
  @Override
  public void dispose() {
      if (disposable != null) {
          disposable.dispose();
      }
  }
  

  回复于 2024-05-03T17:36:17+08:00