RxJava2：使用 Flowable 和 zipWith

Question

我正在尝试使以下代码对我有用，但有问题，这是一个 sn-p：

private void myMethod() {
Flowable.fromIterable(cache)
.zipWith(this::doesExist, (record, exist) -> {
    // do stuff
    return true;
}).subscrib();
}

private Flowable<Boolean> doesExist(CacheRecord record) {
    // Do something
    return Flowable.just(true);
}

这不编译，知道吗？

更新： 关于以下 sn-p 的任何想法：

Flowable.fromIterable(m_cache) // 
.flatMapCompletable(cachedStation -> { 
    return Single.zip(Single.just(cachedStation), doesIssueExist(cachedStation), (record, exist) -> { 
    System.out.println(cachedStation + ", " + exist); 
    return true; 
}).toCompletable(); 
}).subscribe();

Answer 1

您的doesExist 方法需要CacheRecord 作为参数。但是您给this::doesExist 的方法参考发送了一个Subscriber<? super Object> 的实例，这就是显示不兼容类型错误的原因。

您的方法的扩展形式如下。

  private void myMethod() {
    Flowable.fromIterable(cache)
            .zipWith(new Publisher<Object>() {
                @Override
                public void subscribe(Subscriber<? super Object> s) {
                    doesExist(s);
                }
            }, (record, exist) -> {
                // do stuff
                return true;
            }).subscribe();
 }

这里，第一个参数给zipWith

  new Publisher<Object>() {
        @Override
         public void subscribe(Subscriber<? super Object> s) {
                doesExist(s);
            }
   }

是您将其缩写为this::doesExist

如您所见，zipWith 需要第一个参数Publisher，并且您已经创建了一个匿名发布者，并且在subscribe 方法中您通过发送Subscriber<? super Object> s 来调用doesExist(s)，即不是所需的类型。您的方法引用语句this::doesExist 完全执行上述操作，这就是编译器显示incompatible type 错误的原因。

如果您尝试使用doesExist 方法返回的flowable 来zip Flowable，您可以通过传递有效的CacheRecord 对象直接调用它，无需方法引用，如下所示

Flowable.fromIterable(cache)
        .zipWith(doesExist(anotherCache), (record, exist) -> {
            // do stuff
            return true;
        }).subscribe();

注意：请参阅method reference 了解更多信息

更新：如果您尝试将fromIterable 发出的项目传递给doesExist 方法并得到boolean 和cacheRecord 的组合结果，那么

如下创建一个持有者类

class CacheRecordResult {
    CacheRecord cacheRecord;
    boolean isExist;

    public CacheRecordResult(CacheRecord cacheRecord, boolean isExist) {
        this.cacheRecord = cacheRecord;
        this.isExist = isExist;
    }
   }

然后订阅CacheRecordResult如下

  private void myMethod() {
    Flowable.fromIterable(cache)
          .flatMap(cacheRecord -> doesExist(cacheRecord)
                  .map(exist -> new CacheRecordResult(cacheRecord, exist)))
          .subscribe(cacheRecordResult -> {
                CacheRecord cacheRecord = cacheRecordResult.cacheRecord;
                boolean isExist = cacheRecordResult.isExist;
            });
  }