经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Scala » 查看文章
深圳scala-meetup-20180902（2）- Future vs Task and ReaderMonad依赖注入
来源：cnblogs　　作者：雪川大虫　　时间：2018/9/25 20:22:00　　对本文有异议

  在对上一次3月份的scala-meetup里我曾分享了关于Future在函数组合中的问题及如何用Monix.Task来替代。具体分析可以查阅这篇博文。在上篇示范里我们使用了Future来实现某种non-blocking数据库操作，现在可以用Task替换Future部分：

  1.   class KVStore[K,V] {
  2.     private val kvs = new ConcurrentHashMap[K,V]()
  3.     def create(k: K, v: V): Task[Unit] = Task.delay(kvs.putIfAbsent(k,v))
  4.     def read(k: K): Task[Option[V]] = Task.delay(Option(kvs.get(k)))
  5.     def update(k: K, v: V): Task[Unit] = Task.delay(kvs.put(k,v))
  6.     def delete(k: K): Task[Boolean] = Task.delay(kvs.remove(k) != null)
  7.   }

Task是一个真正的Monad，我们可以放心的用来实现函数组合：

  1.   type FoodName = String
  2.   type Quantity = Int
  3.   type FoodStore = KVStore[String,Int]
  5.   def addFood(food: FoodName, qty: Quantity)(implicit fs: FoodStore): Task[Quantity] = for {
  6.     current <- fs.read(food)
  7.     newQty = current.map(cq => cq + qty).getOrElse(qty)
  8.     _ <- fs.update(food,newQty)
  9.   } yield newQty
  11.   def takeFood(food: FoodName, qty: Quantity)(implicit fs: FoodStore): Task[Quantity] = for {
  12.     current <- fs.read(food)
  13.     cq = current.getOrElse(0)
  14.     taken = Math.min(cq,qty)
  15.     left = cq - taken
  16.     _ <- if(left > 0) fs.update(food,left) else fs.delete(food)
  17.   } yield taken
  20.   def cookSauce(qty: Quantity)(get: (FoodName,Quantity) => Task[Quantity],
  21.                                put: (FoodName,Quantity) => Task[Quantity]): Task[Quantity] = for {
  22.     tomato <- get("Tomato",qty)
  23.     vaggies <- get("Veggies",qty)
  24.     _ <- get("Galic",10)
  25.     sauceQ = tomato/2 + vaggies * 3 / 2
  26.     _ <- put("Sauce",sauceQ)
  27.   } yield sauceQ
  29.   def cookPasta(qty: Quantity)(get: (FoodName,Quantity) => Task[Quantity],
  30.                                put: (FoodName,Quantity) => Task[Quantity]): Task[Quantity] = for {
  31.     pasta <- get("Pasta", qty)
  32.     sauce <- get("Sauce", qty)
  33.     _ <- get("Spice", 3)
  34.     portions = Math.min(pasta, sauce)
  35.     _ <- put("Meal", portions)
  36.   } yield portions

跟上次我们使用Future时的方式没有两样。值得研究的是如何获取Task运算结果，及如何更精确的控制Task运算如取消运行中的Task：

  1.   implicit val refridge = new FoodStore
  3.   val shopping: Task[Unit] = for {
  4.     _ <- addFood("Tomato",10)
  5.     _ <- addFood("Veggies",15)
  6.     _ <- addFood("Garlic", 42)
  7.     _ <- addFood("Spice", 100)
  8.     _ <- addFood("Pasta", 6)
  9.   } yield()
  11.   val cooking: Task[Quantity] = for {
  12.     _ <- shopping
  13.     sauce <- cookSauce(10)(takeFood(_,_),addFood(_,_))
  14.     meals <- cookPasta(10)(takeFood(_,_),addFood(_,_))
  15.   } yield meals
  17.  import scala.util._
  18.   import monix.execution.Scheduler.Implicits.global
  20.   val cancellableCooking = Cooking.runOnComplete { result =>
  21.     result match {
  22.       case Success(meals) => println(s"we have $meals pasta meals for the day.")
  23.       case Failure(err) => println(s"cooking trouble: ${err.getMessage}")
  24.     }
  25.   }
  27.   global.scheduleOnce(1 second) {
  28.     println(s"its taking too long, cancelling cooking ...")
  29.     cancellableCooking.cancel()
  30.   }

在上面例子里的addFood,takeFood函数中都有个fs:FoodStore参数。这样做可以使函数更加通用，可以对用不同方式实施的FoodStore进行操作。这里FoodStore就是函数的依赖，我们是通过函数参数来传递这个依赖的。重新组织一下代码使这种关系更明显：

  1.   class Refridge {
  2.     def addFood(food: FoodName, qty: Quantity): FoodStore => Task[Quantity] = { foodStore =>
  3.       for {
  4.         current <- foodStore.read(food)
  5.         newQty = current.map(c => c + qty).getOrElse(qty)
  6.         _ <- foodStore.update(food, newQty)
  7.       } yield newQty
  8.     }
  10.     def takeFood(food: FoodName, qty: Quantity): FoodStore => Task[Quantity] = { foodStore =>
  11.       for {
  12.         current <- foodStore.read(food)
  13.         cq = current.getOrElse(0)
  14.         taken = Math.min(cq, qty)
  15.         left = cq - taken
  16.         _ <- if (left > 0) foodStore.update(food, left) else foodStore.delete(food)
  17.       } yield taken
  18.     }
  20.   }

现在我们用一个函数类型的结果来代表依赖注入。这样做的好处是简化了函数主体，彻底把依赖与函数进行了分割，使用函数时不必考虑依赖。

scala的函数式组件库cats提供了一个Kleisli类型，reader monad就是从它推导出来的：

  1.  final case class Kleisli[M[_], A, B](run: A => M[B]) { self =>
  2.  ...
  3.  trait KleisliFunctions {
  4.    /**Construct a Kleisli from a Function1 */
  5.    def kleisli[M[_], A, B](f: A => M[B]): Kleisli[M, A, B] = Kleisli(f)
  6.  …
  7.   def >=>[C](k: Kleisli[M, B, C])(implicit b: Bind[M]): Kleisli[M, A, C] =  
  8.           kleisli((a: A) => b.bind(this(a))(k.run))
  9.  …
  10.  Kleisli的用途就是进行函数的转换
  11.  // (A=>M[B]) >=> (B=>M[C]) >=> (C=>M[D]) = M[D]

实际上Kleisli就是ReaderT：

  1.  type ReaderT[F[_], E, A] = Kleisli[F, E, A]
  2.  val ReaderT = Kleisli
  3.  val reader = ReaderT[F,B,A](A => F[B]) 
  4.  val readerTask = ReaderT[Task,B,A](A => Task[B])
  5.  val injection = ReaderT { foodStore => Task.delay { foodStore.takeFood } }
  6.  val food = injection.run(db) // run(kvs), run(dbConfig) …

这段代码里我们也针对上面的例子示范了ReaderT的用法。现在我们可以把例子改成下面这样：

  1.   type FoodName = String
  2.   type Quantity = Int
  3.   type FoodStore = KVStore[String,Int]
  5.   class Refridge {
  6.     def addFood(food: FoodName, qty: Quantity): ReaderT[Task,FoodStore,Quantity] = ReaderT{ foodStore =>
  7.       for {
  8.         current <- foodStore.read(food)
  9.         newQty = current.map(c => c + qty).getOrElse(qty)
  10.         _ <- foodStore.update(food, newQty)
  11.       } yield newQty
  12.     }
  14.     def takeFood(food: FoodName, qty: Quantity): ReaderT[Task,FoodStore,Quantity] = ReaderT{ foodStore =>
  15.       for {
  16.         current <- foodStore.read(food)
  17.         cq = current.getOrElse(0)
  18.         taken = Math.min(cq, qty)
  19.         left = cq - taken
  20.         _ <- if (left > 0) foodStore.update(food, left) else foodStore.delete(food)
  21.       } yield taken
  22.     }
  24.   }

ReaderT[F[_],E,A]就是ReaderT[Task,FoodStore,Quantity]. FoodStore是注入的依赖，ReaderT.run返回Task：

  1.   val cooking: ReaderT[Task,FoodStore,Quantity] = for {
  2.     _ <- shopping
  3.     sauce <- cooker.cookSauce(10)
  4.     pasta <- cooker.cookPasta(10)
  5.   } yield pasta
  7.   import scala.concurrent.duration._
  8.   import scala.util._
  9.   import monix.execution.Scheduler.Implicits.global
  10.   val timedCooking = cooking.run(foodStore).timeoutTo(1 seconds, Task.raiseError( new RuntimeException(
  11.     "oh no, take too long to cook ...")))
  12.   val cancellableCooking = timedCooking.runOnComplete { result =>
  13.     result match {
  14.       case Success(meals) => println(s"we have $meals specials for the day.")
  15.       case Failure(exception) => println(s"kitchen problem! ${exception.getMessage}")
  16.     }
  17.   }
  18.   global.scheduleOnce(3 seconds) {
  19.     println("3 seconds passed，cancelling ...")
  20.     cancellableCooking.cancel()
  21.   }

我们知道cooking是个ReaderT，用run(foodStore)来注入依赖foodStore。那么如果我们还有一个kvStore或者jdbcDB,mongoDB可以直接用run(kvStore), run(jdbcDB), run(mongoDB) ... 返回的结果都是Task。

 

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Scala热门文章

  • scala (3) Function 和 Method
  • flink 两个datastream实现left_join的两种方法
  • scala case class + BeanProperty+fastjson 快速实现类的序列化与反序列化
  • Scala解析MDF文件
  • flink 如何实现对watermark 的checkpoint,防止数据复写
  • Spark笔记(一):错误总结
  • 踩坑 scala 打包上传 找不到类
  • akka-typed(2) - typed-actor交流方式和交流协议
  • Scala(二)：元组、数组、映射
  • scala 命名规范

    • Scala推荐文章

  • Scala中Array和List的区别说明
  • alpakka-kafka(5)-kafka集群配置与分布式应用部署
  • alpakka-kafka(2)-consumer
  • alpakka-kafka(1)-producer
  • akka-typed(1) - actor生命周期管理
  • akka-typed(2) - typed-actor交流方式和交流协议
  • akka-typed(3) - PersistentActor has EventSourcedBehavior
  • akka-typed(4) - EventSourcedBehavior in action
  • kka-typed(5) - cluster：集群节点状态监视
  • akka-typed(6) - cluster：group router, cluster-load-balancing

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号