我们如何使用 Monix Tasks 从一系列任务中写入单个输出流(可以并行运行):例如,我有 N 个任务可以并行运行并从中获得一些响应HTTP call/server,
,我正在写回响应到文件流(因为有限制内存),但我stream closed error,
想知道是否有办法使用 Monix 任务或任何其他 API 来实现这一点
java.io.IOException: Stream closed
示例代码片段:
import java.io.{FileOutputStream, OutputStream, OutputStreamWriter}
import monix.eval.Task
import monix.execution.ExecutionModel.AlwaysAsyncExecution
import monix.execution.Scheduler
import scala.util.Random
object TaskTest extends App {
implicit val scheduler = Scheduler(
AlwaysAsyncExecution
)
val list = List(1 to 100)
val filePath = System.currentTimeMillis() + "_test.txt"
val outputStreamWriter = new OutputStreamWriter(new FileOutputStream(filePath))
val futureTask = autoClose(outputStreamWriter) {
outputStream =>
val totalCount = SeqOfTasks(outputStream)
totalCount.map(_.sum)
}
futureTask.runToFuture.onComplete {
case scala.util.Success(value) =>
println(s"fetched a total of $value")
case scala.util.Failure(exception) =>
println("there is some error occurred ")
println(exception)
}
//close this stream when all tasks completed.
def autoClose[A <: AutoCloseable, B](resource: A)(code: A ⇒ B): B = {
try code(resource)
finally resource.close()
}
def SeqOfTasks(outPutStream: OutputStreamWriter): Task[Seq[Int]] = {
val list = List(0 to 10: _*)
Task.wanderUnordered(list)(l => makeHttpCall(outPutStream, l))
}
// each taks will fetch some data and write to single out put stream that is shared by other tasks.
def makeHttpCall(outputStream: OutputStreamWriter, value: Int): Task[Int] = {
Task {
val result = getDummyValue(value)
val bytes = result.mkString("\t") + "\n"
outputStream.write(bytes)
//return some value
result.size
}
}
//return a dummy response for http
def getDummyValue(n: Int): Seq[Int] = {
println(s"dummy value : $n")
Seq.fill(n)(Random.nextInt)
}