Akka(5): ConsistentHashing Router - 可选定Routee的任务分配模式

    上一篇讨论里我们介绍了几种任务分配(Routing)模式。Akka提供的几种现成智能化Routing模式大多数是通过对用户屏蔽具体的运算Routee选择方式来简化Router使用,提高智能程度,所以我们提到Router的运算是一种无序的运算,消息之间绝对不容许任何形式的依赖,因为向Router发送的消息可能在任何Routee上运算。但是,如果我们能够把运算任务按照任务的类型分配给专门负责处理此等类型任务的Routee,那么我们就可以充分利用Routing模式所带来的运算拓展能力来提高整体运算效率。Akka的ConsistentHashingRouter就是为了满足这样的需求而提供的。ConsistentHashingRouter是通过消息的特征来分辨消息类型,然后自动构建和管理处理各种类型消息的Routees。当然,这就要求系统的消息必须具备预先设定的特征,使ConsistentHashingRouter可以正确分辨并分配给指定的Routee去运算。如果我们确定只有一个Routee负责处理一种类型消息的话,甚至可以在这个Routee中维护某种状态。我们可以设计一个场景来示范ConsistentHashingRouter的应用:模拟一个多货币的存钱盒,分n次随意从盒里取出钱币然后统计各种货币的总额。这个场景中的特征很明显:就是货币种类了,我们把抽出的货币按币种、金额合成消息发给ConsistentHashingRouter。例子里的Routee应该是按照币种由Router自动构建的,维护各种货币当前总额作为内部状态。向ConsistentHashingRouter发送的消息被分配给相应币种的Routee去登记更新货币当前总额。这个统计金额的Routee可以如下定义:

import akka.actor._

val currencies = List("RMB","USD","EUR","JPY","GBP","DEM","HKD","FRF","CHF")

object MoneyCounter {
  sealed trait Counting
  case class OneHand(cur: String, amt: Double) extends Counting
  case class ReportTotal(cur: String) extends Counting
}
class MoneyCounter extends Actor with ActorLogging {
  import MoneyCounter._
  var currency: String = "RMB"
  var amount: Double = 0

  override def receive: Receive = {
    case OneHand(cur,amt) => 
      currency = cur
      amount += amt
      log.info(s"${self.path.name} received one hand of $amt$cur")
    case ReportTotal(_) => 
      log.info(s"${self.path.name} has a total of $amount$currency")
  }
}

MoneyCounter支持两项功能:一是统计某种货币收到的总额,二是按指令汇报当前总额。我们在前一篇讨论里了解到如果MoneyCounter是Routee类型,那它们应该被视为具相同功能的Actor。而且用户无法分辨或者直接面对某个特定的Routee。任何MoneyCounter都可以收到一手任何货币,不同的货币金额相加结果是错误的。所以我们要用Akka提供的ConsistentHashingRouter来解决这个问题。ConsistentHashingRouter的主要特点是能够分辨消息类型,然后按照消息类型对应到选定的Routee。在我们上面的例子里每个Routee负责一种货币,这样就可以保证每个Routee里的金额总数都是正确的了。ConsistentHashingRouter有三种分辨消息的方法:

1、定义ConsistentHashingRouter的hashMapping函数:这是个PartialFunction[Any,Any],如下:

object HashingRouter extends App {
 import MoneyCounter._

  val currencies = List("RMB","USD","EUR","JPY","GBP","DEM","HKD","FRF","CHF")

  val routerSystem = ActorSystem("routerSystem")

  def mcHashMapping: PartialFunction[Any,Any] = {
    case OneHand(cur,_) => cur
    case ReportTotal(cur) => cur
  }

  val router = routerSystem.actorOf(ConsistentHashingPool(
    nrOfInstances = 5,hashMapping = mcHashMapping,virtualNodesFactor = 2)
  .props(MoneyCounter.props),name = "moneyCounter" )


  router ! OneHand("RMB",10.00)
  router ! OneHand("USD",10.00)
  router ! OneHand("HKD",10.00)
  router ! OneHand("RMB",10.00)
  router ! OneHand("CHF",10.00)

  router ! ReportTotal("RMB")
  router ! ReportTotal("USD")

  scala.io.StdIn.readLine()

  routerSystem.terminate()
}

我们在定义router时直接把mcHashingMapping传到ConsistentHashingPool的构建器里就行了。特别要注意nrOfInstances,这个参数必须比消息类型的数量大才行,否则Router会错误引导消息。测试运算结果显示如下:

INFO] [06/05/2017 15:20:09.334] [routerSystem-akka.actor.default-dispatcher-6] [akka://routerSystem/user/moneyCounter/$e] $e received one hand of 10.0RMB
[INFO] [06/05/2017 15:20:09.334] [routerSystem-akka.actor.default-dispatcher-3] [akka://routerSystem/user/moneyCounter/$b] $b received one hand of 10.0USD
[INFO] [06/05/2017 15:20:09.334] [routerSystem-akka.actor.default-dispatcher-7] [akka://routerSystem/user/moneyCounter/$d] $d received one hand of 10.0CHF
[INFO] [06/05/2017 15:20:09.334] [routerSystem-akka.actor.default-dispatcher-2] [akka://routerSystem/user/moneyCounter/$a] $a received one hand of 10.0HKD
[INFO] [06/05/2017 15:20:09.334] [routerSystem-akka.actor.default-dispatcher-6] [akka://routerSystem/user/moneyCounter/$e] $e received one hand of 10.0RMB
[INFO] [06/05/2017 15:20:09.337] [routerSystem-akka.actor.default-dispatcher-2] [akka://routerSystem/user/moneyCounter/$b] $b has a total of 10.0USD
[INFO] [06/05/2017 15:20:09.337] [routerSystem-akka.actor.default-dispatcher-6] [akka://routerSystem/user/moneyCounter/$e] $e has a total of 20.0RMB

Router自动调用了e,b,d,a4个Routees,并且能把消息引导到正确的Routee。

2、可以让消息继承ConsistentHashable,如此我们要在消息里实现函数constentHashKey, 如下:

object MoneyCounter {
  sealed class Counting(cur: String) extends ConsistentHashable {
    override def consistentHashKey: Any = cur
  }
  case class OneHand(cur: String, amt: Double) extends Counting(cur)
  case class ReportTotal(cur: String) extends Counting(cur)
  def props = Props(new MoneyCounter)
}

现在消息都是ConsistentHashable类型的了。构建新的Router来测试效果:

  val router = routerSystem.actorOf(ConsistentHashingPool(
    nrOfInstances = 5, virtualNodesFactor = 2).props(
    MoneyCounter.props),name = "moneyCounter")


  router ! OneHand("RMB",10.00)
  router ! OneHand("USD",10.00)
  router ! OneHand("HKD",10.00)
  router ! OneHand("RMB",10.00)
  router ! OneHand("CHF",10.00)

  router ! ReportTotal("RMB")
  router ! ReportTotal("USD")

运算结果同样正确:

[INFO] [06/05/2017 15:36:29.746] [routerSystem-akka.actor.default-dispatcher-7] [akka://routerSystem/user/moneyCounter/$e] $e received one hand of 10.0RMB
[INFO] [06/05/2017 15:36:29.746] [routerSystem-akka.actor.default-dispatcher-5] [akka://routerSystem/user/moneyCounter/$b] $b received one hand of 10.0USD
[INFO] [06/05/2017 15:36:29.746] [routerSystem-akka.actor.default-dispatcher-6] [akka://routerSystem/user/moneyCounter/$a] $a received one hand of 10.0HKD
[INFO] [06/05/2017 15:36:29.746] [routerSystem-akka.actor.default-dispatcher-4] [akka://routerSystem/user/moneyCounter/$d] $d received one hand of 10.0CHF
[INFO] [06/05/2017 15:36:29.746] [routerSystem-akka.actor.default-dispatcher-7] [akka://routerSystem/user/moneyCounter/$e] $e received one hand of 10.0RMB
[INFO] [06/05/2017 15:36:29.749] [routerSystem-akka.actor.default-dispatcher-7] [akka://routerSystem/user/moneyCounter/$e] $e has a total of 20.0RMB
[INFO] [06/05/2017 15:36:29.749] [routerSystem-akka.actor.default-dispatcher-4] [akka://routerSystem/user/moneyCounter/$b] $b has a total of 10.0USD

3、直接把消息包在ConsistentHashableEnvelope里:

  router ! ConsistentHashableEnvelope(message = OneHand("RMB",23.00),hashKey = "RMB")

这种方式需要用户手工指定Routee,如果用这种方式,我们其实不必用Router,直接把消息传给专职的Actor就行了。

看来还是第二种方法比较合适。因为比起第一种方法多了类型安全和与Router的松散耦合。下面就是一个用第二种方法的完整示范源代码:

import akka.actor._
import akka.routing.ConsistentHashingRouter.{ConsistentHashMapping, ConsistentHashable, ConsistentHashableEnvelope}
import akka.routing._


object MoneyCounter {
  sealed class Counting(cur: String) extends ConsistentHashable {
    override def consistentHashKey: Any = cur
  }
  case class OneHand(cur: String, amt: Double) extends Counting(cur)
  case class ReportTotal(cur: String) extends Counting(cur)
  def props = Props(new MoneyCounter)
}
class MoneyCounter extends Actor with ActorLogging {
  import MoneyCounter._
  var currency: String = "RMB"
  var amount: Double = 0

  override def receive: Receive = {
    case OneHand(cur,amt) =>
      currency = cur
      amount += amt
      log.info(s"${self.path.name} received one hand of $amt$cur")
    case ReportTotal(_) =>
      log.info(s"${self.path.name} has a total of $amount$currency")
  }
}
object HashingRouter extends App {
 import MoneyCounter._
  import scala.util.Random

  val currencies = List("RMB","USD","EUR","JPY","GBP","DEM","HKD","FRF","CHF")

  val routerSystem = ActorSystem("routerSystem")

  val router = routerSystem.actorOf(ConsistentHashingPool(
    nrOfInstances = currencies.size+1, virtualNodesFactor = 2).props(
    MoneyCounter.props),name = "moneyCounter")

  (1 to 20).toList foreach (_ => router ! OneHand(
    currencies(Random.nextInt(currencies.size-1))
  ,Random.nextInt(100) * 1.00))

  currencies foreach (c => router ! ReportTotal(c))

  scala.io.StdIn.readLine()

  routerSystem.terminate()
}

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Java成神之路

Java企业微信开发_02_通讯录同步

       登录企业微信—>管理工具—>通讯录同步助手—>开启“API接口同步”  ; 开启后,即可看到通讯录密钥,也可设置通讯录API的权限:读取或者编辑通...

2232
来自专栏码匠的流水账

聊聊storm supervisor的启动

storm-core-1.2.2-sources.jar!/org/apache/storm/daemon/supervisor/Supervisor.java

973
来自专栏逆向技术

PE文件格式详解,第一讲,DOS头文件格式

           PE文件格式详解,第一讲,DOS头文件格式 今天讲解PE文件格式的DOS头文件格式 首先我们要理解,什么是文件格式,我们常说的EXE可执行...

1895
来自专栏知晓程序

开发 | 技术高人如何开发小程序?他们用这套方法

792
来自专栏熊二哥

Java基础组件快速入门

最近需要上线很多新的JAVA项目,然而很多JAVA的相关库都不太熟悉,项目实现起来遇到了不小阻力,熬了好几天夜。现在手头的工作基本完成了,因此打算好好来归纳下j...

2136
来自专栏进击的程序猿

orm 系列 之 常用设计模式 The Repository Pattern

本文是orm系列的第一篇,内容来自github上的一个Markdown,清晰的讲述了一些数据库设计上常用的设计模式,并且阐述了orm是什么?

1113
来自专栏Java帮帮-微信公众号-技术文章全总结

Web-第二十二天 Web商城实战二【悟空教程】

`cname` varchar(20) DEFAULT NULL, #分类名称

1114
来自专栏码匠的流水账

JCTools简介

JCTools是一款对jdk并发数据结构进行增强的并发工具,主要提供了map以及queue的增强数据结构。原来netty还是自己写的MpscLinkedQueu...

471
来自专栏菩提树下的杨过

JAVA CDI 学习(1) - @Inject基本用法

CDI(Contexts and Dependency Injection 上下文依赖注入),是JAVA官方提供的依赖注入实现,可用于Dynamic Web M...

972
来自专栏蓝天

Redis源码笔记-初步

Redis代码优美,注释也很到位,阅读起来会赏心悦目,大大降低了理解门槛。由于redis单线程几乎完成所有工作,整体逻辑是相当复杂的,涉及了太多状态,作者的...

1502

扫码关注云+社区