开始使用 RxJava 和 sqlite

我正在学习 RxJava，并且已经阅读了几篇文章和视频。我对 RxJava 所提供的功能感到非常敬畏，所以我认为目前已经了解了响应式编程的全部内容。Dan Lew 的本教程 http://blog.danlew.net/2014/09/15/grokking-rxjava-part-1/帮助我在一定程度上了解了使用 RxJava 的基础知识。但我发现，我越认为自己已经完全理解了 RxJava，我的脑海中就会冒出越多的问题。

由于我是一个在编写代码时学到最多的人，因此这就是我正在尝试做的事情。我已经有一个使用 SQLite 的应用程序。我的应用程序有 7 个饼图，它从数据库游标获取数据。每个饼图从光标处获取不同行的数据（浮点值），并根据其数据生成图表。

我现在想做的是，我想使用 RxJava 从数据库中检索数据并填充图表。但我不知道 RxJava 应该如何与数据库交互。我需要使用任何额外的库来执行此操作吗？因为我发现rxjava-jdbc https://github.com/davidmoten/rxjava-jdbc and SqlBrite https://corner.squareup.com/2015/02/sqlbrite-reactive-sqlite-for-android.html我认为这应该对我有帮助，但我真的需要它们吗？我还找到了一个SO 中的问题 https://stackoverflow.com/questions/30075663/flatten-observableobservablecursor-to-observablecursor它似乎没有使用任何额外的库（当然除了 RxJava）并且他似乎使用了 ContentObservable 但我仍然没有完全理解它。此外，回答他问题的人甚至没有使用 ContentObservable。我刚刚知道Android有自己的Observable类..它与RxJava一起使用完全兼容吗？如果不是，为什么同名？

有人请帮助我...

=================================================== ======

针对 GreyBeardedGeek 的解决方案如下：

这是我的 Callable，它从 sqlite 检索数据

public class GetIncome implements Callable<Map<String,String>> {
    public static final String FILE_NAME = "CGetIncome";

    Context myContext;
    int year, month;

    public GetIncome(Context context, int getYear, int getMonth){
        myContext = context;
        year = getYear;
        month = getMonth;
    }

    @Override
    public Map<String,String> call() throws Exception {
        Map<String,String> output = new HashMap<>();
        JarControl jc = new JarControl(myContext);
        JSONObject getIncome = jc.getIncome(year,month,0,0);

        output.put("necessities",getIncome.getString("necessities"));
        output.put("savings",getIncome.getString("savings"));

        return output;
    }
}

以下是我尝试使用主要活动中的 Callable 的方式：

Callable<Map<String,String>> getIncome = new GetIncome(getContext(),2015,9);

现在我不知道如何将 getIncome 可调用放入 Observables 中。你提到 rx 中的 fromCallables 应该这样做..但实际上如何做到这一点？根据本教程展示了如何使用 Observable.from http://blog.danlew.net/2014/09/22/grokking-rxjava-part-2/，所以我希望能够通过执行 Observable.fromCallable 来使用 Callables，但显然该函数不存在。快速谷歌搜索表明 .fromCallable 应该通过执行类似 Async.fromCallable() 之类的操作来使用，但由于我对 RxJava 很陌生，所以我只能了解如何根据如何创建可调用教程 http://blog.danlew.net/2014/09/22/grokking-rxjava-part-2/显示...

对于我对 RxJava 的极度不懂，深表歉意

我确信还会有很多其他意见，但到目前为止，我已经采取了几种不同的方法：

• 如果您想要相对少量的数据（并且您可以保证它始终具有已知的大小上限），那么：

  • 创建一个同步方法来访问数据库，获取游标，迭代游标以创建包含数据的数据结构（例如列表），然后返回它。
  • 创建一个调用该方法的 Callable
  • 使用 rx.fromCallable 创建一个 Observable

• 如果您需要无限数量的行，请执行与上述类似的操作，但返回游标。

编辑：示例代码：

private Map<String, String> getDataFromDatabase() { 
   Map<String, String> result = new HashMap<>(); 
   // do whatever you need to (i.e. db query, cursor) to fill it in 
   return result; 
} 

private Callable<Map<String, String>> getData() { 
   return new Callable() { 
      public Map<String, String> call() { 
        return getDataFromDatabase(); 
    } 
}

// in utility class  RxUtil.java 
public static <T> Observable<T> makeObservable(final Callable<T> func) {
        return Observable.create(
                new Observable.OnSubscribe<T>() {
                    @Override
                    public void call(Subscriber<? super T> subscriber) {
                        try {
                            T observed = func.call();
                            if (observed != null) { // to make defaultIfEmpty work
                                subscriber.onNext(observed);
                            }
                            subscriber.onCompleted();
                        } catch (Exception ex) {
                            subscriber.onError(ex);
                        }
                    }
                });
    }


// and finally, 
public Observable<Map<String, String>> getDataObservable() {
   return RxUtil.makeObservable(getData());
}