如何使用 RxJava 和 Kotlin 进行 groupBy 和收集？答案

【问题标题】：How to do a groupBy and collect using RxJava and Kotlin?如何使用 RxJava 和 Kotlin 进行 groupBy 和收集？
【发布时间】：2016-10-31 02:48:00
【问题描述】：

我有Observable<Rates>，Rate 只是一个简单的对象：

Rate(val value:String){}
Rates(val rates: List<Rate>)

我想将 Observable<Rates> 更改为 Observable<HashMap<String,Long>。

例如，对于费率Rates(arrayOf(Rate("1"),Rate("2"), Rate("3"),Rate("3"), Rate("2"),Rate("2")))，我期望结果：

(1 -> 1)
(2 -> 3)
(3 -> 2)
(4 -> 0)
(5 -> 0)

我开始创造类似的东西：

service.getRates()
        .flatMap {it-> Observable.from(it.rates) }
        .filter { !it.value.isNullOrEmpty() }
        .groupBy {it -> it.value}
        .collect({ HashMap<String,Long>()}, { b, t -> b.put(t.key, t.count???)}

但我被困在这里，我不知道计算所有值？如果没有 4 个中的 5 个，我不知道如何添加空值 (0)。有没有办法使用 rx 来做到这一点？

【问题讨论】：

标签： rx-java reactive-programming kotlin rx-kotlin

【解决方案1】：

请查看代码中的 cmets 以获得您问题的答案。

import rx.Observable

fun main(args: Array<String>) {
    val service = Service()

    // This adds all keys with each key mapped to zero
    val referenceKeyCounts = Observable
        .just("1", "2", "3", "4", "5")
        .map { it to 0 }

    val keyCountsFromService = service.getRates()
        .flatMap { Observable.from(it.rates) }
        .filter { !it.value.isNullOrEmpty() }
        .map { it.value to 1 } // map each occurrence of key to 1

    Observable.concat(referenceKeyCounts, keyCountsFromService)
        .groupBy { it.first }
        .flatMap { group ->  // this converts GroupedObservable to final values
            group.reduce(0, { acc, pair -> acc + pair.second }) // add instead of counting
                .map { group.key to it }
        }
        .subscribe(::println)

}

class Service {
    fun getRates(): Observable<Rates> = Observable.just(Rates(listOf(
        Rate("1"), Rate("2"), Rate("3"), Rate("3"), Rate("2"), Rate("2")
    )))
}

class Rate(val value: String)

class Rates(val rates: List<Rate>)

【讨论】：

【解决方案2】：

诀窍是在GroupedObservable 上使用count，因为它只在源可观察对象完成时发出单个值：

从那里开始：

rates
  .flatMap { Observable.from(it.rates) }
  .filter { !it.value.isNullOrEmpty() }
  .groupBy { it.value }
  .flatMap { group -> group.count().map { group.key to it } } // list "1"->1, "2"->3, ...
  .mergeWith(Observable.from((1..5).map { it.toString() to 0 })) // defaults "4"->0
  .reduce(mutableMapOf<String, Int>()) { acc, cur ->
      acc.apply {
        val (key, count) = cur  
        this[key] = (this[key] ?: 0) + count // add counts
      }
  }.subscribe { countedRates -> 
    println(countedRates)
}

【讨论】：

【解决方案3】：

我认为它更多是关于函数式编程，而不是与 RxJava 相关的问题

从Rates -> Map<String,Int> 实现一个映射函数。

技巧：将两个Pair<String,Int>列表合并成一个Map<String,Int>

val ratesToMapWithEmptyValues: (Rates) -> Map<String, Int> = { source ->

  //TODO: Just for demo
  val validRatesValue = arrayOf("1","2","3","4","5")

  mapOf(
    *validRatesValue.map { it to 0 }.toTypedArray(),
    *source.rates.groupBy(Rate::value).mapValues { it.value.size }.toList().toTypedArray()
  )

}

应用Observable.map中的函数

service.getRates()
        .map(ratesToMapWithEmptyValues)

【讨论】：