RX vs Theads vs Executors中mutlitheading的性能

Question

我正在用 Kotlin 编写后端应用程序。

为了加快速度，我目前依靠服务器上的 RxKotlin 来并行执行 IO 任务，例如数据库调用和 API 调用。代码通常是这样的。

val singleResult1 = Single.fromCallable{
  database.get(....)
}.io()

val singleResult2 = Single.fromCallable{
  database.update(....)
}.io()

Single.zip(singleResult1, singleResult2){ result1: Result1, result2: Result2 ->
    ....
}
.flatMap{
  //other RX calls
}
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.computation())
.blockingGet()

但是，由于不能真正处理多个事件（只是单个事件），Rx 感觉有点混乱，只是添加了一堆样板（如果我想返回一个空值，它也会导致复杂化，有时可能会弄乱堆栈跟踪)

我正在考虑删除 Rx 并使用Executors（或线程）来代替并行性。这里是否有任何性能方面的考虑？

例如我在想什么：

fun <T> waitAll(tasks: List<Callable<T>>, threadCount: Int = -1): List<T> {
    val threads = if (threadCount == -1) tasks.size else threadCount
    val executor = Executors.newFixedThreadPool(threads)
    val results = executor.invokeAll(tasks).map {
        it.get()
    }
    executor.shutdown()
    return results
}

并像这样使用它：

waitAll(listOf(callable1, callable2))

或者也许使用常规线程并加入它们？

threads.forEach{
   it.start()
}
threads.forEach{
   it.join()
}

或者为什么不流？

listOf(callable1,callable2)
.parallelStream()
.map{it.call()}
.collect(Collectors.toList())

Answer 1

Java Executor 服务使用线程，而 RxKotlin 使用 ExecutorServices。所以所有这些在背景中都是一样的。区别在于软件架构。因此，如果您选择与您的代码集成的最佳架构，它将发挥最佳效果并正确完成工作。简单就是最好的。

如果您有一个基于事件或基于 observable 的架构，并且您正在尝试实现一个新库来处理基于事件的操作或作业，您可能会编写关于作业分离或计时的错误步骤。使用 RxKotlin，不要再发明轮子了。

如果您的工作与事件或可观察模式无关，而您只需要执行并行作业，请使用 Executor 服务。RxKotlin 将过度工程化。当你在这种情况下使用 RxKotlin 时，你需要做更多的事情。

所以我认为问题不在于这种情况下的速度，而在于架构。

Answer 2

Schedulers.io()KotlinRx 本身使用执行器，但它在（无界）和Schedulers.computation()（受核心数量限制）下有两个预先创建的线程池，并且不会像您建议的代码那样每次都启动一个新线程池。您显然也可以手动执行此操作：

private val executor = Executors.newCachedThreadPool() // equivalent to io()

fun <T> waitAll(tasks: List<Callable<T>>): List<T> {
    return executor.invokeAll(tasks).map {
        it.get()
    }
}

一般来说，这应该比为每个任务创建一个线程更好，允许重用现有线程，但这取决于您的使用情况。

协程（IMO）更适合处理后台/ui线程通信（即Android等）

协程对此是否有用很大程度上取决于您的任务中的内容。阻塞 API（例如 JDBC）？他们不是。异步的（例如 Retrofit）？他们是。