指定任务 fs2 流的处理时间 Scala

Question

我正在尝试处理一些任务.任务所需的时间可能会因元素而异。例如，存储和获取元素 1 从队列中取出的时间可能需要11秒，而 2 可能需要30秒...

我已经尝试使用一个计时器，但仍然，我得到的entryTime = ExitTime我想知道我错过了什么。

这里我已经尝试了。

import cats.effect.{ExitCode, IO, IOApp, Timer}
import fs2._
import fs2.concurrent.Queue
import scala.concurrent.duration._
import scala.util.Random

class Tst(q1: Queue[IO, (Double, String, String)])(implicit timer: Timer[IO]) {

  import core.Processing._

  def storeInQueue: Stream[IO, Unit] = {

    Stream(1, 2, 3)
      .covary[IO]
      .evalTap(n => IO.delay(println(s"Pushing $n to Queue")))
      .map { n =>
        val entryTime = currentTimeNow
        (n.toDouble, "Service", entryTime)
      //  timer.sleep(Random.between(10, 30).seconds) I have tried adding it here but the same result
       }
      .through(q1.enqueue)  
  }

      def getFromQueue: Stream[IO, Unit] = {
        timer.sleep(Random.between(10, 30).seconds)
        q1.dequeue
          .map { n =>
            val exitTime = currentTimeNow
            (n._1, "Service", n._3, exitTime)
          }
          .evalMap(n => IO.delay(println(s"Pulling from queue $n")))
      }
    }

    object Five2 extends IOApp {

      override def run(args: List[String]): IO[ExitCode] = {
        val program = for {
          q <- Queue.bounded[IO, (Double, String, String)](10)
          b = new Tst(q)
          _ <- b.storeInQueue.compile.drain.start

          _ <- b.getFromQueue.compile.drain

        } yield ()
        program.as(ExitCode.Success)
      }
    }

currentTimeNow方法是用.NET技术给出的。

def currentTimeNow: String = {
    val format = new SimpleDateFormat("dd-MM-yy hh:mm:ss")
    format.format(Calendar.getInstance().getTime())
  }

Answer 1

得到的帮助 咬文嚼字 : 这就是答案 你的做法有几个问题。首先，你的 timer.sleep call 是被忽略的。因为它返回一个IO，而IO不会做任何事情，除非它被评估为. 所以你会希望它在你的流的管道中执行，你可以用类似于

q1.dequeue.evalTap(_ => timer.sleep(Random.between(10, 30).seconds))).map { n => … other things … }

第二件事是，我不知道你的currentTimeNow函数是如何工作的，但通常在函数栈中获取当前时间是一个有效的操作，所以使用定时器，你更多的是像这样获取当前时间。

Stream(1, 2, 3)
      .covary[IO]
      .evalTap(n => IO.delay(println(s"Pushing $n to Queue")))
      .evalMap { n =>
        timer.clock.realTime(java.util.concurrent.TimeUnit.MILLISECONDS)
        .map(entryTime => (n.toDouble, “Service”, entryTime))
      }
      .through(q1.enqueue)