使用 FS2 的对象池模式

Question

我正在尝试查看在FS2中实现对象池模式的最佳方法是什么。

假设我们有以下MyPrinter定义：

class MyPrinter {
  import scala.util.Random.nextInt
  Thread.sleep(5000 + nextInt(1000))
  def doStuff(s: String): Unit = {
    println(s)
    Thread.sleep(1000 + nextInt(1000))
  }
  def releaseResources(): Unit =
    println("Releasing resources")
}

制作由n打印机池支持的Stream[Task, MyPrinter]的最佳方法是什么？当流结束时，应通过调用 releaseResources 正确释放所有底层资源。

奖励问题：如果打印机由于某种原因而关闭，是否可以在池中创建一台新打印机？

Answer 1

不确定我是否得到了这个问题，但是这个怎么样

implicit val S = Strategy.fromFixedDaemonPool(10, "pooling")
val queue = new LinkedBlockingDeque[MyPrinter]()
queue.add(new MyPrinter)
queue.add(new MyPrinter)
Stream.repeatEval(Task.delay(queue.take()))
  .map(p => try p.doStuff("test") finally {
    p.releaseResources()
    queue.put(p)
  })
  .take(10)
  .runLog
  .unsafeRun()

队列可以替换为 https://commons.apache.org/proper/commons-pool/

更新：

如果要同时处理每个"资源"：

concurrent.join(10)(
  Stream
    .repeatEval(Task.delay(queue.take()))
    .map(p => Stream.eval(Task.delay(p.doStuff("test"))
    .map(_ => p /* done with this resource */)))
).map(p => { p.releaseResources(); queue.put(p) /* release resource */})
 .take(10).runLog.unsafeRun()