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 条评论
登录 后参与评论

相关文章

来自专栏我和未来有约会

Silverlight网络寻奇 at 090428

都是Silverlight做的游戏 http://www.silverarcade.com/Games Introducing Html Utilities f...

2994
来自专栏前端说吧

canvas练习 - 圆

652
来自专栏ios 技术积累

ios view 圆角

765
来自专栏静晴轩

TOP 100 大前端超棒精选列表

正值如今这信息爆炸的年代,如何能从中汲取精华,于有限时间内,成为更高效的学习者,从而在激烈的竞争中更具优势,是当下每个人或企业都该思虑的问题;先前创立的 Web...

1623
来自专栏三好码农的三亩自留地

Android-打造一个简单通用的Material加载LoadingView

因为网络请求的时间一般是未知的,所以我们一般都是用一个循环的圆圈指示器来提示用户,如下图。

1573
来自专栏张善友的专栏

ASP.NET开发人员需要学习ASP.NET MVC么?

InfoQ上有篇文章ASP.NET开发人员需要学习ASP.NET MVC么? 文章已经覆盖了绝大多数对于学习/使用ASP.NET MVC的观点。 我觉得ASP....

1899
来自专栏张善友的专栏

Silverlight 2 和 Deep Zoom

Steve Clayton 发布了一篇关于 Silverlight 2 演示的文章,其中有提到一项叫作 Deep Zoom 的技术,现部署在 http://me...

1776
来自专栏转载gongluck的CSDN博客

cocos2dx 2048

#include "GameScene.h" #include "CardSprite.h" #include "ScoreBoard.h" #include ...

3658
来自专栏NetCore

Fluent NHibernate 之旅 导航篇

ORM大家都非常熟悉了吧,我相信也有很多朋友正在用自己或者一些公开的框架,而最常用而且强大的,非Hibernate了(Net中为NHibernate),网上的文...

1867
来自专栏我和未来有约会

44 Amazing Silverlight 2.0 Screencasts

Silverlight - Hello World Silverlight - Anatomy of an Application Silverlight - ...

2019

扫码关注云+社区