FunDA(8)- Static Source:保证资源使用安全 - Resource Safety

   我们在前面用了许多章节来讨论如何把数据从后台数据库中搬到内存,然后进行逐行操作运算。我们选定的解决方案是把后台数据转换成内存中的数据流。无论在打开数据库表或从数据库读取数据等环节都涉及到对数据库表这项资源的安全使用:我们最起码要保证在完成使用或者使用中途出现错误异常退出时能释放占用的资源。谈到资源使用安全,不得不想到函数式编程通用的那个bracket函数,fs2同样提供了这个函数:

def bracket[F[_],R,A](r: F[R])(use: R => Stream[F,A], release: R => F[Unit]): Stream[F,A] = Stream.mk {
    StreamCore.acquire(r, release andThen (Free.eval)) flatMap { case (_, r) => use(r).get }
  }

这个函数的入参数r,use,release都涉及到了资源占用处理:r一般是打开文件或者库表操作,use是资源使用如读取数据过程,release 顾名思义就是正常完成资源使用后的资源释放清理过程。函数bracket能保证这些过程的正确引用。

我们用几个例子来分析一下这个函数的功能:

val s = Stream.bracket(Task.delay(throw new Exception("Oh no!")))(
  _ => Stream(1,2,3) ++ Stream.fail(new Exception("boom!")) ++ Stream(3,4),
  _ => Task.delay(println("normal end")))         
s.runLog.unsafeRun                                //> java.lang.Exception: Oh no!
                                                  //|     at demo.ws.streams$$anonfun$main$1$$anonfun$1.apply(demo.ws.streams.scal
                                                  //| a:4)
                                                  //|     at demo.ws.streams$$anonfun$main$1$$anonfun$1.apply(demo.ws.streams.scal
                                                  //| a:4)

在上面这个例子里我们人为在两个地方制造了异常。我们可以用onError来截获这些异常: 

val s1 = s.map(_.toString).onError {e => Stream.emit(e.getMessage)}
                                                  
s1.runLog.unsafeRun                               //> res0: Vector[String] = Vector(Oh no!)

必须用toString转换了Stream元素类型后才能把截获的异常信息放进Stream。注意release未调用,因为资源还没有被占用。但是如果除了释放资源外还有其它清理工作的话,我们可以用onFinalize来确保一定可以调用清理程序:

val s5 = s1.onFinalize(Task.delay{println("finally end!")})
 
s5.runLog.unsafeRun                               //> finally end!
                                                  //| res1: Vector[String] = Vector(Oh no!)

如果在使用资源中间出现异常会怎样?

val s3 = Stream.bracket(Task.delay())(
  _ => Stream(1,2,3) ++ Stream.fail(new Exception("boom!")) ++ Stream(3,4),
  _ => Task.delay(println("normal end"))) 
val s4 = s3.map(_.toString).onError {e => Stream.emit(e.getMessage)}
         .onFinalize(Task.delay{println("finally end!")})

s4.runLog.unsafeRun                               //> normal end
                                                  //| finally end!
                                                  //| res2: Vector[String] = Vector(1, 2, 3, boom!)

返回结果res2正确记录了出错地点,而且所有清理过程都得到运行。当然,我们可以不用动Stream元素类型,用attempt:

val s6 = s3.attempt.onError {e => Stream.emit(e.getMessage)}
         .onFinalize(Task.delay{println("finally end!")})
 
s6.runLog.unsafeRun                               //> normal end
                                                  //| finally end!
 //| res3: Vector[Object] = Vector(Right(1), Right(2), Right(3), Left(java.lang.Exception: boom!))

我们在前面FunDA(1)里讨论过运算slick Query Action run返回结果类型是Future[Iterable[ROW]]。Slick获取数据的方式是一次性读入内存,所以本期标题提到的Static-Source就是指这样的一个内存中的集合。那么我们就可以不必考虑开启并占用数据库表这项操作了。我们只需要用FunDA DataRowType.getTypedRow函数获取了Iterable[ROW]结果后直接传给bracket就行了。现在最重要的是如何把Seq[ROW]转换成Stream[F[_],ROW]。我们可以用Seq的fold函数来构建Stream: 

val data = Seq(1,2,3,4)                           //> data  : Seq[Int] = List(1, 2, 3, 4)
val s8 = data.foldLeft(Stream[Task,Int]())((s,a) => s ++ Stream.emit(a))
def log[A](prompt: String): Pipe[Task,A,A] =
    _.evalMap {row => Task.delay{ println(s"$prompt> $row"); row }}
                                                  //> log: [A](prompt: String)fs2.Pipe[fs2.Task,A,A]

s8.through(log("")).run.unsafeRun                 //> > 1
                                                  //| > 2
                                                  //| > 3
                                                  //| > 4

表面上看好像没什么问题,但仔细分析:Seq[ROW]可以是个超大的集合,而foldLeft是个递归函数,无论是否尾递归都有可能造成堆栈溢出错误(StackOverflowError)。看来还是用freemonad,它可以把每步运算都存放在内存结构里,可以在固定的堆栈空间运算。下面的函数用fs2.Pull类型结构可以把Seq[ROW]转换成Stream[F[_],ROW]:

 def pullSeq[ROW](h: Seq[ROW]): Pull[Task, ROW, Unit] = {
    val it = h.iterator
    def go(it: Iterator[ROW]): Pull[Task, ROW, Unit] = for {
      res <- Pull.eval(Task.delay({ if (it.hasNext) Some(it.next()) else None }))
      next <- res.fold[Pull[Task, ROW, Unit]](Pull.done)(o => Pull.output1(o) >> go(it))
    } yield next
    go(it)
  }                                                  
 def streamSeq[ROW](h: Seq[ROW]): Stream[Task, ROW] =
    pullSeq(h).close

虽然go是个递归函数,但因为Pull是个freemonad,每个flapMap循环(>>)会把新的Iterable it状态存放在heap内存里。由于每个步骤都是存放在内存结构里的,而运算这些步骤的模式是靠下游拖动逐步运算的,也就是说按下游拖动每次产生一个元素。pullSeq返回Pull,Pull.close >>> Stream,这就是streamSeq函数的作用了。现在我们可以直接用bracket来安全构建Stream:

 val s9 = Stream.bracket(Task.delay(data))(streamSeq, _ => Task.delay())                                                
 s9.through(log("")).run.unsafeRun               //> > 1
                                                  //| > 2
                                                  //| > 3
                                                  //| > 4

现在可以放心了。但我们的目的是为大众编程人员提供一个最低门槛的工具库,他们不需要了解Task, onError,onFinalize。。。我们必须把bracket函数使用方式搞得更直白点,让用户可以更容易调用:

  type FDAStream[A] = Stream[Task,A]
  implicit val strategy = Strategy.fromFixedDaemonPool(4)
                                                  //> strategy  : fs2.Strategy = Strategy

  def fda_staticSource[ROW](acquirer: => Seq[ROW],
                            releaser: => Unit = (),
                            errhandler: Throwable => FDAStream[ROW] = null,
                            finalizer: => Unit = ()): FDAStream[ROW] = {
     val s = Stream.bracket(Task(acquirer))(r => streamSeq(r), r => Task(releaser))
     if (errhandler != null)
       s.onError(errhandler).onFinalize(Task.delay(finalizer))
     else
       s.onFinalize(Task.delay(finalizer))
  }                                               //> fda_staticSource: [ROW](acquirer: => Seq[ROW], releaser: => Unit, errhandle
                                                  //| r: Throwable => demo.ws.streams.FDAStream[ROW], finalizer: => Unit)demo.ws.
                                                  //| streams.FDAStream[ROW]

如果完整调用fda_staticSource可以如下这样:

  val s10 = fda_staticSource(data,
     println("endofuse"), e => { println(e.getMessage);Stream.emit(-99) },
     println("finallyend")) 
  s10.through(log("")).run.unsafeRun              //> > 1
                                                  //| > 2
                                                  //| > 3
                                                  //| > 4
                                                  //| endofuse
                                                  //| finallyend

最简单直接的方式如下:

  val s11 = fda_staticSource(acquirer = data) 
  s11.through(log("")).run.unsafeRun              //> > 1
                                                  //| > 2
                                                  //| > 3
                                                  //| > 4

又或者带异常处理过程的调用方法:

  val s12 = fda_staticSource(acquirer = data, errhandler = {e => println(e.getMessage);Stream()})
 
  s12.through(log("")).run.unsafeRun              //> > 1
                                                  //| > 2
                                                  //| > 3
                                                  //| > 4

下面是这次讨论示范的源代码:

import fs2._
object streams {
val s = Stream.bracket(Task.delay(throw new Exception("Oh no!")))(
  _ => Stream(1,2,3) ++ Stream.fail(new Exception("boom!")) ++ Stream(3,4),
  _ => Task.delay(println("normal end")))
//s.runLog.unsafeRun
val s1 = s.map(_.toString).onError {e => Stream.emit(e.getMessage)}
s1.runLog.unsafeRun
val s5 = s1.onFinalize(Task.delay{println("finally end!")})
s5.runLog.unsafeRun

val s3 = Stream.bracket(Task.delay())(
  _ => Stream(1,2,3) ++ Stream.fail(new Exception("boom!")) ++ Stream(3,4),
  _ => Task.delay(println("normal end")))
val s4 = s3.map(_.toString).onError {e => Stream.emit(e.getMessage)}
         .onFinalize(Task.delay{println("finally end!")})
s4.runLog.unsafeRun
val s6 = s3.attempt.onError {e => Stream.emit(e.getMessage)}
         .onFinalize(Task.delay{println("finally end!")})

s6.runLog.unsafeRun
                                                  
val data = Seq(1,2,3,4)
val s8 = data.foldLeft(Stream[Task,Int]())((s,a) => s ++ Stream.emit(a))
def log[A](prompt: String): Pipe[Task,A,A] =
    _.evalMap {row => Task.delay{ println(s"$prompt> $row"); row }}

s8.through(log("")).run.unsafeRun
                                              
  def pullSeq[ROW](h: Seq[ROW]): Pull[Task, ROW, Unit] = {
    val it = h.iterator
    def go(it: Iterator[ROW]): Pull[Task, ROW, Unit] = for {
      res <- Pull.eval(Task.delay({ if (it.hasNext) Some(it.next()) else None }))
      next <- res.fold[Pull[Task, ROW, Unit]](Pull.done)(o => Pull.output1(o) >> go(it))
    } yield next
    go(it)
  }
  def streamSeq[ROW](h: Seq[ROW]): Stream[Task, ROW] =
    pullSeq(h).close
  val s9 = Stream.bracket(Task.delay(data))(streamSeq, _ => Task.delay())
  s9.through(log("")).run.unsafeRun
  
  type FDAStream[A] = Stream[Task,A]
  implicit val strategy = Strategy.fromFixedDaemonPool(4)

  def fda_staticSource[ROW](acquirer: => Seq[ROW],
                            releaser: => Unit = (),
                            errhandler: Throwable => FDAStream[ROW] = null,
                            finalizer: => Unit = ()): FDAStream[ROW] = {
     val s = Stream.bracket(Task(acquirer))(r => streamSeq(r), r => Task(releaser))
     if (errhandler != null)
       s.onError(errhandler).onFinalize(Task.delay(finalizer))
     else
       s.onFinalize(Task.delay(finalizer))
  }
  val s10 = fda_staticSource(data,
     println("endofuse"), e => { println(e.getMessage);Stream.emit(-99) },
     println("finallyend"))
  s10.through(log("")).run.unsafeRun
   val s11 = fda_staticSource(acquirer = data)
   s11.through(log("")).run.unsafeRun
   val s12 = fda_staticSource(acquirer = data, errhandler = {e => println(e.getMessage);Stream()})
   s12.through(log("")).run.unsafeRun
 
 }

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏吉浦迅科技

DAY11:阅读CUDA异步并发执行中的Event和同步调用

743
来自专栏用户画像

逻辑覆盖 测试题

(2)  简述什么是测试用例。白盒测试的动态测试要根据程序的控制结构设计测试用例简述其原则。

492
来自专栏Ryan Miao

Java8-理解Collector

2724
来自专栏lzj_learn_note

阿里ARouter拦截器使用及源码解析(二)

关于ARouter基本跳转的用法以及源码解析在上篇文章阿里阿里ARouter使用及源码解析(一)已经有过分析,有不清楚的同学可以去看看。本篇文章主要是关于ARo...

692
来自专栏逸鹏说道

Python3 与 C# 扩展之~基础衍生

在线编程: https://mybinder.org/v2/gh/lotapp/BaseCode/master

1867
来自专栏一个会写诗的程序员的博客

bootstrap-table 前端分页,刷新事件代码实例

参考文档: http://bootstrap-table.wenzhixin.net.cn/

1001
来自专栏技术墨客

Java数据校验详解

一个健壮的系统都要对外部提交的数据进行完整性、合法性的校验。即使开发一个不面对最终用户的工具包,也需要对传入的数据进行缜密的校验来防止引发底层难以追踪的问题。各...

952
来自专栏10km的专栏

java:基于volatile和Thread Local Storage的双重检查锁定实现延迟初始化

总在同一个地方栽坑里是不可原谅的,以本文做个记录,以防下次再犯。 下面这段很简单的基于双重检查锁定(Double-checked locking)实现的延迟...

2157
来自专栏颇忒脱的技术博客

Spring MVC异步处理简介

本文讲到的所有特性皆是基于Servlet 3.0 Async Processing的,不是基于Servlet 3.1 Async IO的。

603
来自专栏Albert陈凯

2018-08-02 你应该知道的高性能无锁队列Disruptor你应该知道的高性能无锁队列Disruptor1.何为队列2.jdk中的队列3.Disruptor4.Log4j中的Disruptor最

https://juejin.im/post/5b5f10d65188251ad06b78e3

962

扫码关注云+社区