如何从 rxjava 中的 Single Observable 获取值？答案

【问题标题】：How to get the value from Single Observable in rxjava?如何从 rxjava 中的 Single Observable 获取值？
【发布时间】：2019-06-14 15:50:28
【问题描述】：

private void getAccount(String accountName,String password) {
    Log.i("ACCOUNT_", "any message"); //this is executed
    Single.fromCallable(() -> {
        String account = accountName;
        Log.i("ACCOUNT_", account); //not executed
        return account;
    }).flatMap((accountName) ->{
        return accountRepository.findAccount(accountName);
    }).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe((List<Account> accounts) -> {
                Account accountModel = accounts.get(0);//not executed
                Log.i("ACCOUNT_", accountModel.getName());//not executed
            },throwable -> {
               Log.i("ACCOUNT_", "BAD EROR");//not executed
      });

}

我更新了代码注释，当我调用getAcount() 方法时哪些部分没有被执行。可能是什么原因？

【问题讨论】：

Log.i("ACCOUNT_", accountModel.getName()); 未执行
subscribe 方法中的所有内容都没有执行
不抛出错误

标签： java android rx-java rx-java2 android-livedata

【解决方案1】：

我几乎可以肯定这与AndroidSchedulers.mainThread() 上的执行有关，你observe 你的Single。请确保该线程正在运行并且在您的代码执行之前没有终止。

很遗憾，我手头没有安卓环境，你的代码也不完整。但作为演示，我使用标准 Java 的模拟方法重写了您的代码。看起来像这样：

package com.dpopov.rxjava.stackoverflow;

import io.reactivex.Single;
import io.reactivex.SingleObserver;
import io.reactivex.SingleSource;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

import java.util.Collections;
import java.util.List;
import java.util.Set;

public class ObservableNotExecuting {

    public static void main(String[] args) throws InterruptedException {
        getAccount("accName", "accPass");

        Thread.sleep(5000);
    }

    private static void getAccount(String accountName, String password) {
        System.out.println("ACCOUNT_" + "any message"); // this is executed

        Single.fromCallable(() -> {
            String account = accountName;
            System.out.println("ACCOUNT_" + account); // not executed
            return account;
        })
                .flatMap((Function<String, SingleSource<?>>) ObservableNotExecuting::findAccount)
                .subscribeOn(Schedulers.io())
//                .observeOn(Schedulers.mainThread())
//                .observeOn(Schedulers.mainThread())
                .subscribe(
                        accounts -> {
//                        accountObject -> {
//                            Account accountModel = accounts.get(0); // not executed
//                            System.out.println("ACCOUNT_" + accounts.name); // not executed
                            final Account account = ((List<Account>) accounts).get(0);
                            System.out.println("account: " + account.name);
                        },
                        throwable -> {
                            System.out.println("ACCOUNT_" + "BAD ERROR"); // not executed
                        }
                );
    }

    private static SingleSource<List<Account>> findAccount(final String accountName) {
        return new Single<List<Account>>() {
            @Override
            protected void subscribeActual(final SingleObserver<? super List<Account>> observer) {
                final Account account = new Account(accountName);
                final List<Account> accountsList = Collections.singletonList(account);
                observer.onSuccess(accountsList);
            }
        };
    }

    static class Account {
        public final String name;

        public Account(final String name) {
            this.name = name;
        }
    }
}

您可以尝试在本地执行此操作。要点是：如果删除#main 中的Thread.sleep 行，则仅将第一行写入输出，因为main 线程在#getAccount 返回之后立即终止，在任何Observable 逻辑开始执行之前.

因此，在执行 RxJava 代码时，看看你的 Android UI 线程是如何运行的。

【讨论】：