我有一个Process[Task, A]
,我需要在流的每个A
上运行一个函数A => B
,其运行时间范围从瞬时到很长,以产生一个Process[Task, B]
。
问题在于,我希望在ExecutionContext
中尽快处理每个A
,并在获得结果后立即传递结果,而不管收到A
的顺序如何。
import java.util.concurrent.{TimeUnit, Executors}
import scala.concurrent.ExecutionContext
import scalaz.stream._
import scalaz.concurrent.Task
object Test extends App {
val executor = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(8))
Process.range(0, 100).flatMap { i =>
Process.eval(Task.apply {
if(i % 2 == 0) Thread.sleep(500)
i
}(executor))
}.to(io.printStreamSink(System.out)(_ println _))
.run.run
executor.shutdown()
executor.awaitTermination(10, TimeUnit.MINUTES)
}
答案是使用通道。下面是更新后的代码,它似乎完全符合我的要求:
import java.util.concurrent.{TimeUnit, Executors}
import scala.concurrent.ExecutionContext
import scalaz.stream._
import scalaz.concurrent.Task
object Test extends App {
val executor = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(8))
val chan = channel.lift[Task, Int, Int] { i => Task {
if(i % 2 == 0) Thread.sleep(500)
i
}}
merge.mergeN(8)(Process.range(0, 100).zipWith(chan)((i, f) => Process.eval(f(i))))
.to(io.printStreamSink(System.out)(_ println _)).run.run
executor.shutdown()
executor.awaitTermination(10, TimeUnit.MINUTES)
}