在 RxJava 2 / RxAndroid 2 中使用“地图”的问题

Question

我正在尝试学习适用于 Android 的 RXJAVA。部分是有意义的，我仍然对很多其他部分感到困惑，但是，有一段时间我希望这一切都会变得更有意义。

目前我在使用“地图”功能时遇到问题。我收到一个错误，但不知道如何解决它。

在分享我的代码之前，我先解释一下我的理解。 在一个简单的水平.. Observable - 发出数据的代码。 观察者 - 处理发出的数据的代码。 Map - 接收类型 A 的数据并将其返回已处理或作为类型 B 的代码。

所以，考虑到这一点：

在 gradle 中我有：

compile 'io.reactivex.rxjava2:rxjava:2.0.1' 
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

如果我有：

    //declaration at top of file
    private Observable<Integer> myIntObservable;
    private Observer<Integer> myIntObserver;
    private Observer<String> myStringObserver;
    private Observable<String> myStringObservable;

    //usage in a function
    myIntObserver = new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onNext(Integer value) {
            Toast.makeText(getApplicationContext(), "" + value, Toast.LENGTH_SHORT).show();
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {
            Toast.makeText(getApplicationContext(), "Int Observer Async Complete", Toast.LENGTH_SHORT).show();
        }
    };


    //Connect my Observable to the observer.
    myIntObservable.observeOn(Schedulers.io());
    myIntObservable.subscribeOn(AndroidSchedulers.mainThread());
    myIntObservable.subscribe(myIntObserver);

这一切都很好......我的地图用法类似......

我想做的是使用同样的 observable 返回一个 int，然后使用映射代码来代替返回一个字符串...

因此：

    myStringObservable
            .observeOn(Schedulers.io())
            .subscribeOn(AndroidSchedulers.mainThread())
            .map(new Function<Integer, String>() {
                @Override
                public String apply(Integer query){
                    return "String Observable result == " + query;
                }
            });

现在，我有两个问题：

a) 我收到的构建错误是： Error:(179, 17) 错误：Observable 类中的方法映射不能应用于给定类型； 必需：功能 发现：> 原因：无法推断类型变量 R （参数不匹配；> 不能转换为函数） 其中 R,T 是类型变量： R 扩展了在方法 map(Function) 中声明的 Object T 扩展了 Observable 类中声明的 Object

我相信这本质上是在告诉我这些类型不适合我的用法，但是我看不清楚...如何解决这个问题。

b) 我在上面发布的地图代码没有将可观察对象连接到它需要观察的内容......因此，我应该在地图命令之前添加订阅行吗？

因此，我尝试了这个..

public void setupAsyncSubscription(){

    myIntObservable
            .observeOn(Schedulers.io())
            .subscribeOn(AndroidSchedulers.mainThread())
            .subscribe(myIntObserver)
            .map(new Function<Integer, String>() {
                @Override
                public String apply(Integer query){
                    return "String Observable result == " + query;
                }
            });

}

虽然这消除了“a”中详述的错误，但它却给我提供了以下错误：

Error:(180, 17) error: void 不能被取消引用（这指向 'map' 调用）

最后，我还可以看到，我从“地图”函数“返回”的内容根本没有被处理……我不清楚如何处理。我觉得在这种情况下我应该使用 .subscribe 调用？？

我“认为”我正在慢慢解决问题.

一如既往，感谢您的帮助。

Answer 1

这里有多个问题。有——一一：

//Connect my Observable to the observer.
myIntObservable.observeOn(Schedulers.io());
myIntObservable.subscribeOn(AndroidSchedulers.mainThread());
myIntObservable.subscribe(myIntObserver);

上面的代码不会像你想象的那样工作。

运算符observeOn、subscribeOn 并非旨在更改内部可观察状态。他们正在返回具有所需行为的新可观察对象。

要在 io() 线程上完成观察并在 mainThread() 上订阅您的 observable，您需要更改代码：

//Connect my Observable to the observer.
myIntObservable = myIntObservable.observeOn(Schedulers.io());
myIntObservable = myIntObservable.subscribeOn(AndroidSchedulers.mainThread());
myIntObservable.subscribe(myIntObserver);

或使用（首选）链接：

//Connect my Observable to the observer.
myIntObservable
    .observeOn(Schedulers.io());
    .subscribeOn(AndroidSchedulers.mainThread());
    .subscribe(myIntObserver);

对于与您相同的代码，在未更改的 Observable 上调用 .subscribe() 将导致在您调用 .subscribe() 的同一线程上订阅和观察（很可能来自主线程）。

请记住，一旦工作完成，您需要处置 observable。

当涉及到映射问题时 - map() 运算符将一种类型的 Observable<A> 更改为另一种类型的可观察 Observable<B>。

如果您希望得到从 Integer 对象转换而来的 String 对象，您需要使用 - 作为数据源 - 您原来的 myIntObservable：

myStringObservable = myIntObservable
    (...)    
    .map(new Function<Integer, String>() {
        @Override
        public String apply(Integer query){
            return "String Observable result == " + query;
        }
    });

在上面的示例中，myIntObservable 将发出 Integer 对象（正如 .apply(Integer query) 方法中所预期的那样。然后 .map() 运算符将创建另一个 Observable 类型为 Observable<String> 您可以分配给 myStringObservable（或任何你想要的）。

然后，您可以使用myStringObservable 订阅它的事件：

myStringObservable.subscribe(myStringObserver)

同样，请记住在工作完成后处理Observable。

还请注意，您应该：

• .observeOn() 尽快完成当前工作，
• .subscribeOn() 尽可能晚（你不想在主线程上继续 io() 或 computation() 操作，对吧？）。

最后提示：考虑在 RxJava 中使用 lambda。使用所有这些匿名类（new Function() 等）将使您的代码在不久的将来难以阅读。

Answer 2

第一个问题是myStringObservable 发出String 类的对象，但您正在尝试映射需要Integer 类参数的函数。为了实现你想要的，你应该使用myIntObservable。

第二个问题是subscribe 调用会为你返回一个Disposable 对象，之后你就不能再做map。删除subscribe 电话，这应该没问题。另请注意，您的方法签名告诉它返回String，但它不能返回String，它可以返回Observable<String>。