Akka

在实际应用中，集群环境里共用一些数据是不可避免的。我的意思是有些数据可以在任何节点进行共享同步读写，困难的是如何解决更改冲突问题。本来可以通过分布式数据库来实现这样的功能，但使用和维护成本又过高，不值得。分布式数据类型distributed-data （ddata）正是为解决这样的困局而设计的。akka提供了一组CRDT（ConflictFreeReplicatedDataType 免冲突可复制数据类型）和一套管理方法来实现分布式数据在集群中的免冲突共享共用。

akka提供的分布式数据共享管理方案是通过replicator来实现的。replicator就是一种actor, 在集群的每一个节点运行replicator后，各节点相同actor路径（去掉地址信息后）的replicator可以通过gissip协议进行沟通，仿佛连接成一个replicator网络通道。replicator提供一套解决数据更新冲突及数据同步的api。首先，共享数据结构是在各节点的replicator中构建的，数据更新时各节点程序把包嵌共享数据类型指定和对该数据更新方法函数的消息发送给本节点的replicator去更新并通过gossip协议向其它节点的replicator同步，同时解决同步时发生的冲突问题。由于数据是存在于replicator内的，所以数据值的读取同样是通过向本地replicator发送数据读取消息实现的。

replicator作为一个actor，可以通过在.conf文件中定义akka-cluster-ddata-DistributedData扩展来启动，又或者直接通过replicator.prop构建。个人认为直接构建actor会灵活许多，而且可以在一个节点上构建多个replicator，因为不同节点上的replicator是通过actor路径来分群组的。下面是通过replicator.prop构建replicator的示范代码：

1  val replicator = system.actorOf(Replicator.props(
2    ReplicatorSettings(system).withGossipInterval(1.second)), "replicator")

如果使用配置文件中的akka.extension 进行构建：

1akka {
2   extensions = ["akka.cluster.ddata.DistributedData"]
3 ...
4}
5
6val replicator = DistributedData(context.system).replicator

CRDT是某种key,value数据类型。CRDT value主要包括Counter,Flag,Set,Map几种类型，包括：

1/**
2 * Implements a boolean flag CRDT that is initialized to `false` and
3 * can be switched to `true`. `true` wins over `false` in merge.
4 *
5 * This class is immutable, i.e. "modifying" methods return a new instance.
6 */
7final case class Flag(enabled: Boolean) 
8final case class FlagKey(_id: String) 
9
10/**
11 * Implements a 'Growing Counter' CRDT, also called a 'G-Counter'.
12 * A G-Counter is a increment-only counter (inspired by vector clocks) in
13 * which only increment and merge are possible. Incrementing the counter
14 * adds 1 to the count for the current node. Divergent histories are
15 * resolved by taking the maximum count for each node (like a vector
16 * clock merge). The value of the counter is the sum of all node counts.
17 *
18 * This class is immutable, i.e. "modifying" methods return a new instance.
19 */
20final class GCounter
21final case class GCounterKey(_id: String)
22
23/**
24 * Implements a 'Increment/Decrement Counter' CRDT, also called a 'PN-Counter'.
25 * PN-Counters allow the counter to be incremented by tracking the
26 * increments (P) separate from the decrements (N). Both P and N are represented
27 * as two internal [[GCounter]]s. Merge is handled by merging the internal P and N
28 * counters. The value of the counter is the value of the P counter minus
29 * the value of the N counter.
30 *
31 * This class is immutable, i.e. "modifying" methods return a new instance.
32 */
33final class PNCounter 
34final case class PNCounterKey(_id: String)
35
36/**
37 * Implements a 'Add Set' CRDT, also called a 'G-Set'. You can't
38 * remove elements of a G-Set.
39 * A G-Set doesn't accumulate any garbage apart from the elements themselves.
40 * This class is immutable, i.e. "modifying" methods return a new instance.
41 */
42final case class GSet[A]
43final case class GSetKey[A](_id: String)
44
45/**
46 * Implements a 'Observed Remove Set' CRDT, also called a 'OR-Set'.
47 * Elements can be added and removed any number of times. Concurrent add wins
48 * over remove.
49 *
50 * The ORSet has a version vector that is incremented when an element is added to
51 * the set. The `node -> count` pair for that increment is stored against the
52 * element as its "birth dot". Every time the element is re-added to the set,
53 * its "birth dot" is updated to that of the `node -> count` version vector entry
54 * resulting from the add. When an element is removed, we simply drop it, no tombstones.
55 *
56 * When an element exists in replica A and not replica B, is it because A added
57 * it and B has not yet seen that, or that B removed it and A has not yet seen that?
58 * In this implementation we compare the `dot` of the present element to the version vector
59 * in the Set it is absent from. If the element dot is not "seen" by the Set version vector,
60 * that means the other set has yet to see this add, and the item is in the merged
61 * Set. If the Set version vector dominates the dot, that means the other Set has removed this
62 * element already, and the item is not in the merged Set.
63 *
64 * This class is immutable, i.e. "modifying" methods return a new instance.
65 */
66final class ORSet[A]
67final case class ORSetKey[A](_id: String)
68
69/**
70 * Implements a 'Observed Remove Map' CRDT, also called a 'OR-Map'.
71 *
72 * It has similar semantics as an [[ORSet]], but in case of concurrent updates
73 * the values are merged, and must therefore be [[ReplicatedData]] types themselves.
74 *
75 * This class is immutable, i.e. "modifying" methods return a new instance.
76 */
77final class ORMap[A, B <: ReplicatedData]
78final case class ORMapKey[A, B <: ReplicatedData](_id: String)
79
80/**
81 * An immutable multi-map implementation. This class wraps an
82 * [[ORMap]] with an [[ORSet]] for the map's value.
83 *
84 * This class is immutable, i.e. "modifying" methods return a new instance.
85 */
86final class ORMultiMap[A, B]
87final case class ORMultiMapKey[A, B](_id: String)
88
89/**
90 * Map of named counters. Specialized [[ORMap]] with [[PNCounter]] values.
91 *
92 * This class is immutable, i.e. "modifying" methods return a new instance.
93 */
94final class PNCounterMap[A]
95final case class PNCounterMapKey[A](_id: String)

综合统计，akka提供现成的CRDT类型包括：

Counters: GCounter, PNCounter
Sets: GSet, ORSet
Maps: ORMap, ORMultiMap, LWWMap, PNCounterMap
Registers: LWWRegister, Flag

CRDT操作结果也可以通过订阅方式获取。用户发送Subscribe消息给replicator订阅有关Key[A]数据的操作结果：

1 /**
2   * Register a subscriber that will be notified with a [[Changed]] message
3   * when the value of the given `key` is changed. Current value is also
4   * sent as a [[Changed]] message to a new subscriber.
5   *
6   * Subscribers will be notified periodically with the configured `notify-subscribers-interval`,
7   * and it is also possible to send an explicit `FlushChanges` message to
8   * the `Replicator` to notify the subscribers immediately.
9   *
10   * The subscriber will automatically be unregistered if it is terminated.
11   *
12   * If the key is deleted the subscriber is notified with a [[Deleted]]
13   * message.
14   */
15  final case class Subscribe[A <: ReplicatedData](key: Key[A], subscriber: ActorRef) extends ReplicatorMessage
16  /**
17   * Unregister a subscriber.
18   *
19   * @see [[Replicator.Subscribe]]
20   */
21  final case class Unsubscribe[A <: ReplicatedData](key: Key[A], subscriber: ActorRef) extends ReplicatorMessage
22  /**
23   * The data value is retrieved with [[#get]] using the typed key.
24   *
25   * @see [[Replicator.Subscribe]]
26   */
27  final case class Changed[A <: ReplicatedData](key: Key[A])(data: A) extends ReplicatorMessage {
28    /**
29     * The data value, with correct type.
30     * Scala pattern matching cannot infer the type from the `key` parameter.
31     */
32    def get[T <: ReplicatedData](key: Key[T]): T = {
33      require(key == this.key, "wrong key used, must use contained key")
34      data.asInstanceOf[T]
35    }
36
37    /**
38     * The data value. Use [[#get]] to get the fully typed value.
39     */
40    def dataValue: A = data
41  }
42
43  final case class Deleted[A <: ReplicatedData](key: Key[A]) extends NoSerializationVerificationNeeded {
44    override def toString: String = s"Deleted [$key]"
45  }

replicator完成操作后发布topic为Key[A]的Changed, Deleted消息。

分布式数据读写是通过发送消息给本地的replicator来实现的。读写消息包括Update,Get,Delete。读取数据用Get，也可以订阅CRDT的更新状态消息Changed, Deleted。

赋予CRDT复制和免冲突特性的应该是replicator对Update这个消息的处理方式。Update消息的构建代码如下：

1  final case class Update[A <: ReplicatedData](key: Key[A], writeConsistency: WriteConsistency,request: Option[Any])(val modify: Option[A] ⇒ A) 
2extends Command[A] with NoSerializationVerificationNeeded {...}
3
4def apply[A <: ReplicatedData](
5      key: Key[A], initial: A, writeConsistency: WriteConsistency,
6      request: Option[Any] = None)(modify: A ⇒ A): Update[A] =
7      Update(key, writeConsistency, request)(modifyWithInitial(initial, modify))
8
9private def modifyWithInitial[A <: ReplicatedData](initial: A, modify: A ⇒ A): Option[A] ⇒ A = {
10      case Some(data) ⇒ modify(data)
11      case None       ⇒ modify(initial)
12    }

我们看到在Update类型里包嵌了数据标示Key[A]和一个函数modify: Option[A] => A。replicator会用这个modify函数来对CRDT数据A进行转换处理。构建器函数apply还包括了A类型数据的初始值，在第一次引用这个数据时就用initial这个初始值，这个从modifyWithInitial函数和它在apply里的引用可以了解。下面是这个Update消息的使用示范：

1  val timeout = 3.seconds.dilated
2
3  val KeyA = GCounterKey("A")
4  val KeyB = ORSetKey[String]("B")
5  val KeyC = PNCounterMapKey[String]("C")
6  val KeyD = ORMultiMapKey[String, String]("D")
7  val KeyE = ORMapKey[String, GSet[String]]("E")
8
9  replicator ! Update(KeyA, GCounter(), WriteAll(timeout))(_ + 3)
10
11  replicator ! Update(KeyB, ORSet(), WriteAll(timeout))(_ + "a" + "b" + "c")
12
13  replicator ! Update(KeyC, PNCounterMap.empty[String], WriteAll(timeout)) { _ increment "x" increment "y" }
14
15  replicator ! Update(KeyD, ORMultiMap.empty[String, String], WriteAll(timeout)) { _ + ("a" → Set("A")) }
16
17  replicator ! Update(KeyE, ORMap.empty[String, GSet[String]], WriteAll(timeout)) { _ + ("a" → GSet.empty[String].add("A")) }

由于CRDT数据读写是通过消息发送形式实现的，读写结果也是通过消息形式返回的。数据读取返回消息里包嵌了结果数据。下面就是读写返回结果消息类型：

1/*------------------UPDATE STATE MESSAGES -----------*/
2  final case class UpdateSuccess[A <: ReplicatedData](key: Key[A], request: Option[Any])
3    extends UpdateResponse[A] with DeadLetterSuppression
4  sealed abstract class UpdateFailure[A <: ReplicatedData] extends UpdateResponse[A]
5
6  /**
7   * The direct replication of the [[Update]] could not be fulfill according to
8   * the given [[WriteConsistency consistency level]] and
9   * [[WriteConsistency#timeout timeout]].
10   *
11   * The `Update` was still performed locally and possibly replicated to some nodes.
12   * It will eventually be disseminated to other replicas, unless the local replica
13   * crashes before it has been able to communicate with other replicas.
14   */
15  final case class UpdateTimeout[A <: ReplicatedData](key: Key[A], request: Option[Any]) extends UpdateFailure[A]
16  /**
17   * If the `modify` function of the [[Update]] throws an exception the reply message
18   * will be this `ModifyFailure` message. The original exception is included as `cause`.
19   */
20  final case class ModifyFailure[A <: ReplicatedData](key: Key[A], errorMessage: String, cause: Throwable, request: Option[Any])
21    extends UpdateFailure[A] {
22    override def toString: String = s"ModifyFailure [$key]: $errorMessage"
23  }
24  /**
25   * The local store or direct replication of the [[Update]] could not be fulfill according to
26   * the given [[WriteConsistency consistency level]] due to durable store errors. This is
27   * only used for entries that have been configured to be durable.
28   *
29   * The `Update` was still performed in memory locally and possibly replicated to some nodes,
30   * but it might not have been written to durable storage.
31   * It will eventually be disseminated to other replicas, unless the local replica
32   * crashes before it has been able to communicate with other replicas.
33   */
34  final case class StoreFailure[A <: ReplicatedData](key: Key[A], request: Option[Any])
35    extends UpdateFailure[A] with DeleteResponse[A] {
36
37/* ---------------- GET MESSAGES --------*/
38  /**
39   * Reply from `Get`. The data value is retrieved with [[#get]] using the typed key.
40   */
41  final case class GetSuccess[A <: ReplicatedData](key: Key[A], request: Option[Any])(data: A)
42    extends GetResponse[A] with ReplicatorMessage {
43
44    /**
45     * The data value, with correct type.
46     * Scala pattern matching cannot infer the type from the `key` parameter.
47     */
48    def get[T <: ReplicatedData](key: Key[T]): T = {
49      require(key == this.key, "wrong key used, must use contained key")
50      data.asInstanceOf[T]
51    }
52
53    /**
54     * The data value. Use [[#get]] to get the fully typed value.
55     */
56    def dataValue: A = data
57  }
58  final case class NotFound[A <: ReplicatedData](key: Key[A], request: Option[Any])
59    extends GetResponse[A] with ReplicatorMessage
60
61
62/*----------------DELETE MESSAGES ---------*/
63  final case class DeleteSuccess[A <: ReplicatedData](key: Key[A], request: Option[Any]) extends DeleteResponse[A]
64  final case class ReplicationDeleteFailure[A <: ReplicatedData](key: Key[A], request: Option[Any]) extends DeleteResponse[A]
65  final case class DataDeleted[A <: ReplicatedData](key: Key[A], request: Option[Any])
66    extends RuntimeException with NoStackTrace with DeleteResponse[A] {
67    override def toString: String = s"DataDeleted [$key]"
68  }

读取返回消息中定义了数据读取方法def dataValue: A 获取数据，或者用类型方法get(Key[A])指定读取目标。下面是一些数据读取例子：

1val replicator = DistributedData(system).replicator
2val Counter1Key = PNCounterKey("counter1")
3val Set1Key = GSetKey[String]("set1")
4val Set2Key = ORSetKey[String]("set2")
5val ActiveFlagKey = FlagKey("active")
6
7replicator ! Get(Counter1Key, ReadLocal)
8
9val readFrom3 = ReadFrom(n = 3, timeout = 1.second)
10replicator ! Get(Set1Key, readFrom3)
11
12val readMajority = ReadMajority(timeout = 5.seconds)
13replicator ! Get(Set2Key, readMajority)
14
15val readAll = ReadAll(timeout = 5.seconds)
16replicator ! Get(ActiveFlagKey, readAll)
17
18case g @ GetSuccess(Counter1Key, req) ⇒
19  val value = g.get(Counter1Key).value
20case NotFound(Counter1Key, req) ⇒ // key counter1 does not exist
21
22...
23
24case g @ GetSuccess(Set1Key, req) ⇒
25  val elements = g.get(Set1Key).elements
26case GetFailure(Set1Key, req) ⇒
27// read from 3 nodes failed within 1.second
28case NotFound(Set1Key, req)   ⇒ // key set1 does not exist
29
30/*---- return get result to user (sender())  ----*/
31
32  case "get-count" ⇒
33    // incoming request to retrieve current value of the counter
34    replicator ! Get(Counter1Key, readTwo, request = Some(sender()))
35
36  case g @ GetSuccess(Counter1Key, Some(replyTo: ActorRef)) ⇒
37    val value = g.get(Counter1Key).value.longValue
38    replyTo ! value
39  case GetFailure(Counter1Key, Some(replyTo: ActorRef)) ⇒
40    replyTo ! -1L
41  case NotFound(Counter1Key, Some(replyTo: ActorRef)) ⇒
42    replyTo ! 0L

下面是用消息订阅方式获取读写状态的示范：

1  replicator ! Subscribe(DataKey, self)
2
3...
4
5    case c @ Changed(DataKey) ⇒
6      val data = c.get(DataKey)
7      log.info("Current elements: {}", data.elements)

在下面我们做一个例子来示范几种CRDT数据的读写和监控操作：

1object DDataUpdator {
2
3  case object IncCounter
4  case class AddToSet(item: String)
5  case class AddToMap(item: String)
6  case object ReadSet
7  case object ReadMap
8  case object ShutDownDData
9
10
11  val KeyCounter = GCounterKey("counter")
12  val KeySet = ORSetKey[String]("gset")
13  val KeyMap = ORMultiMapKey[Long, String]("ormap")
14
15  val timeout = 300 millis
16  val writeAll = WriteAll(timeout)
17  val readAll = ReadAll(timeout)
18
19
20  def create(port: Int): ActorRef = {
21    val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port = $port")
22      .withFallback(ConfigFactory.load())
23    val system = ActorSystem("DDataSystem",config)
24    system.actorOf(Props[DDataUpdator],s"updator-$port")
25  }
26
27}
28
29class DDataUpdator extends Actor with ActorLogging {
30  import DDataUpdator._
31  implicit val cluster = Cluster(context.system)
32  val replicator = DistributedData(context.system).replicator
33
34
35  replicator ! Subscribe(KeyCounter,self)
36  replicator ! Subscribe(KeySet,self)
37  replicator ! Subscribe(KeyMap,self)
38
39  override def receive: Receive = {
40    case IncCounter =>
41       log.info(s"******* Incrementing counter... *****")
42       replicator ! Update(KeyCounter,GCounter(),writeAll)(_ + 1)
43    case UpdateSuccess(KeyCounter,_) =>
44      log.info(s"********** Counter updated successfully ********")
45    case UpdateTimeout(KeyCounter,_) =>
46      log.info(s"******* Counter update timed out! *****")
47    case ModifyFailure(KeyCounter,msg,err,_) =>
48      log.info(s"******* Counter update failed with error: ${msg} *****")
49    case StoreFailure(KeyCounter,_) =>
50      log.info(s"******* Counter value store failed! *****")
51    case c @ Changed(KeyCounter) ⇒
52      val data = c.get(KeyCounter)
53      log.info("********Current count: {}*******", data.getValue)
54
55
56    case AddToSet(item) =>
57      replicator ! Update(KeySet,ORSet.empty[String],writeAll)(_ + item)
58    case UpdateSuccess(KeySet,_) =>
59      log.info(s"**********Add to ORSet successfully ********")
60    case UpdateTimeout(KeySet,_) =>
61      log.info(s"******* Add to ORSet timed out! *****")
62    case ModifyFailure(KeySet,msg,err,_) =>
63      log.info(s"******* Add to ORSet failed with error: ${msg} *****")
64    case StoreFailure(KeySet,_) =>
65      log.info(s"******* ORSet items store failed! *****")
66    case c @ Changed(KeySet) =>
67      val data = c.get(KeySet)
68      log.info("********Items in ORSet: {}*******", data.elements)
69    case ReadSet =>
70      replicator ! Get(KeySet,readAll)
71    case g @ GetSuccess(KeySet, req) =>
72      val value = g.get(KeySet)
73      log.info("********Current items read in ORSet: {}*******", value.elements)
74    case NotFound(KeySet, req) =>
75      log.info("******No item found in ORSet!!!*******")
76
77
78
79    case AddToMap(item) =>
80       replicator ! Get(KeyCounter,readAll,Some(AddToMap(item)))
81    case g @ GetSuccess(KeyCounter,Some(AddToMap(item))) =>
82      val idx: Long = g.get(KeyCounter).getValue.longValue()
83      log.info(s"*********** got counter=${idx} with item: $item ************")
84      replicator ! Update(KeyMap,ORMultiMap.empty[Long,String],writeAll)(_ + (idx -> Set(item)))
85      replicator ! Update(KeyCounter,GCounter(),writeAll)(_ + 1)
86    case c @ Changed(KeyMap) =>
87      val data = c.get(KeyMap).entries
88      log.info("******** Items in ORMultiMap: {}*******", data)
89    case ReadMap =>
90      replicator ! Get(KeyMap,readAll)
91    case g @ GetSuccess(KeyMap, req) =>
92      val value = g.get(KeyMap)
93      log.info("********Current items read in ORMultiMap: {}*******", value.entries)
94    case NotFound(KeyMap, req) =>
95      log.info("****** No item found in ORMultiMap!!! *******")
96
97
98
99    case ShutDownDData => context.system.terminate()
100
101  }

在这个例子里我们示范了每种CRDT数据的通用操作方法。然后我们再测试一下使用结果：

1object DDataDemo extends App {
2  import DDataUpdator._
3
4  val ud1 = create(2551)
5  val ud2 = create(2552)
6  val ud3 = create(2553)
7  scala.io.StdIn.readLine()
8
9  ud1 ! IncCounter
10  ud2 ! AddToSet("Apple")
11  ud1 ! AddToSet("Orange")
12
13  scala.io.StdIn.readLine()
14
15  ud2 ! IncCounter
16  ud2 ! AddToSet("Pineapple")
17  ud1 ! IncCounter
18  ud1 ! AddToMap("Cat")
19
20  scala.io.StdIn.readLine()
21
22  ud1 ! AddToMap("Dog")
23  ud2 ! AddToMap("Tiger")
24  scala.io.StdIn.readLine()
25
26  ud3 ! ReadSet
27  ud3 ! ReadMap
28  scala.io.StdIn.readLine()
29
30
31   ud1 ! ShutDownDData
32   ud2 ! ShutDownDData
33   ud3 ! ShutDownDData
34}

结果如下：

1[INFO] [12/24/2018 08:33:40.500] [DDataSystem-akka.actor.default-dispatcher-16] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ******* Incrementing counter... *****
2[INFO] [12/24/2018 08:33:40.585] [DDataSystem-akka.actor.default-dispatcher-26] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] **********Add to ORSet successfully ********
3[INFO] [12/24/2018 08:33:40.585] [DDataSystem-akka.actor.default-dispatcher-21] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********** Counter updated successfully ********
4[INFO] [12/24/2018 08:33:40.585] [DDataSystem-akka.actor.default-dispatcher-21] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] **********Add to ORSet successfully ********
5[INFO] [12/24/2018 08:33:40.726] [DDataSystem-akka.actor.default-dispatcher-19] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********Current count: 1*******
6[INFO] [12/24/2018 08:33:40.726] [DDataSystem-akka.actor.default-dispatcher-19] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********Items in ORSet: Set(Orange, Apple)*******
7[INFO] [12/24/2018 08:33:40.775] [DDataSystem-akka.actor.default-dispatcher-21] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ********Items in ORSet: Set(Apple, Orange)*******
8[INFO] [12/24/2018 08:33:40.775] [DDataSystem-akka.actor.default-dispatcher-21] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ********Current count: 1*******
9[INFO] [12/24/2018 08:33:40.829] [DDataSystem-akka.actor.default-dispatcher-23] [akka.tcp://DDataSystem@localhost:2553/user/updator-2553] ********Current count: 1*******
10[INFO] [12/24/2018 08:33:40.829] [DDataSystem-akka.actor.default-dispatcher-23] [akka.tcp://DDataSystem@localhost:2553/user/updator-2553] ********Items in ORSet: Set(Apple, Orange)*******
11
12
13[INFO] [12/24/2018 08:34:19.707] [DDataSystem-akka.actor.default-dispatcher-23] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ******* Incrementing counter... *****
14[INFO] [12/24/2018 08:34:19.707] [DDataSystem-akka.actor.default-dispatcher-19] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ******* Incrementing counter... *****
15[INFO] [12/24/2018 08:34:19.710] [DDataSystem-akka.actor.default-dispatcher-23] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ********** Counter updated successfully ********
16[INFO] [12/24/2018 08:34:19.711] [DDataSystem-akka.actor.default-dispatcher-19] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********** Counter updated successfully ********
17[INFO] [12/24/2018 08:34:19.712] [DDataSystem-akka.actor.default-dispatcher-28] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] **********Add to ORSet successfully ********
18[INFO] [12/24/2018 08:34:19.723] [DDataSystem-akka.actor.default-dispatcher-17] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********Current count: 3*******
19[INFO] [12/24/2018 08:34:19.723] [DDataSystem-akka.actor.default-dispatcher-17] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********Items in ORSet: Set(Orange, Apple, Pineapple)*******
20[INFO] [12/24/2018 08:34:19.733] [DDataSystem-akka.actor.default-dispatcher-19] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] *********** got counter=3 with item: Cat ************
21[INFO] [12/24/2018 08:34:19.767] [DDataSystem-akka.actor.default-dispatcher-21] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********** Counter updated successfully ********
22[INFO] [12/24/2018 08:34:19.772] [DDataSystem-akka.actor.default-dispatcher-19] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ********Current count: 4*******
23[INFO] [12/24/2018 08:34:19.773] [DDataSystem-akka.actor.default-dispatcher-19] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ********Items in ORSet: Set(Apple, Orange, Pineapple)*******
24[INFO] [12/24/2018 08:34:19.774] [DDataSystem-akka.actor.default-dispatcher-19] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ******** Items in ORMultiMap: Map(3 -> Set(Cat))*******
25[INFO] [12/24/2018 08:34:19.828] [DDataSystem-akka.actor.default-dispatcher-17] [akka.tcp://DDataSystem@localhost:2553/user/updator-2553] ********Current count: 4*******
26[INFO] [12/24/2018 08:34:19.828] [DDataSystem-akka.actor.default-dispatcher-17] [akka.tcp://DDataSystem@localhost:2553/user/updator-2553] ********Items in ORSet: Set(Apple, Orange, Pineapple)*******
27[INFO] [12/24/2018 08:34:19.828] [DDataSystem-akka.actor.default-dispatcher-17] [akka.tcp://DDataSystem@localhost:2553/user/updator-2553] ******** Items in ORMultiMap: Map(3 -> Set(Cat))*******
28[INFO] [12/24/2018 08:34:20.222] [DDataSystem-akka.actor.default-dispatcher-21] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ******** Items in ORMultiMap: Map(3 -> Set(Cat))*******
29[INFO] [12/24/2018 08:34:20.223] [DDataSystem-akka.actor.default-dispatcher-21] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********Current count: 4*******
30
31[INFO] [12/24/2018 08:34:45.918] [DDataSystem-akka.actor.default-dispatcher-25] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] *********** got counter=4 with item: Tiger ************
32[INFO] [12/24/2018 08:34:45.919] [DDataSystem-akka.actor.default-dispatcher-16] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] *********** got counter=4 with item: Dog ************
33[INFO] [12/24/2018 08:34:45.920] [DDataSystem-akka.actor.default-dispatcher-15] [akka.tcp://DDataSystem@localhost:2553/user/updator-2553] ********Current items read in ORSet: Set(Apple, Orange, Pineapple)*******
34[INFO] [12/24/2018 08:34:45.922] [DDataSystem-akka.actor.default-dispatcher-22] [akka.tcp://DDataSystem@localhost:2553/user/updator-2553] ********Current items read in ORMultiMap: Map(3 -> Set(Cat))*******
35[INFO] [12/24/2018 08:34:45.925] [DDataSystem-akka.actor.default-dispatcher-21] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********** Counter updated successfully ********
36[INFO] [12/24/2018 08:34:45.926] [DDataSystem-akka.actor.default-dispatcher-27] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ********** Counter updated successfully ********
37[INFO] [12/24/2018 08:34:46.221] [DDataSystem-akka.actor.default-dispatcher-2] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ******** Items in ORMultiMap: Map(4 -> Set(Dog, Tiger), 3 -> Set(Cat))*******
38[INFO] [12/24/2018 08:34:46.221] [DDataSystem-akka.actor.default-dispatcher-2] [akka.tcp://DDataSystem@localhost:2551/user/updator-2551] ********Current count: 6*******
39[INFO] [12/24/2018 08:34:46.272] [DDataSystem-akka.actor.default-dispatcher-27] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ******** Items in ORMultiMap: Map(4 -> Set(Tiger, Dog), 3 -> Set(Cat))*******
40[INFO] [12/24/2018 08:34:46.272] [DDataSystem-akka.actor.default-dispatcher-27] [akka.tcp://DDataSystem@localhost:2552/user/updator-2552] ********Current count: 6*******
41
42
43[INFO] [12/24/2018 08:34:46.326] [DDataSystem-akka.actor.default-dispatcher-22] [akka.tcp://DDataSystem@localhost:2553/user/updator-2553] ******** Items in ORMultiMap: Map(4 -> Set(Dog, Tiger), 3 -> Set(Cat))*******
44[INFO] [12/24/2018 08:34:46.326] [DDataSystem-akka.actor.default-dispatcher-22] [akka.tcp://DDataSystem@localhost:2553/user/updator-2553] ********Current count: 6*******

注意最后一段显示结果是在另一个节点2553上读取其它节点上更新的ORSet和ORMultiMap里面的数据。其中Map(4->set(Dog,Tiger)) 应该是分两次读取了Counter后再更新的。不过由于两次消息发送时间间隔太短，Counter还没来得及更新复制。

下面是这个例子的全部源代码：

build.sbt

1name := "akka-distributed-data"
2
3version := "0.1"
4
5scalaVersion := "2.12.8"
6
7libraryDependencies := Seq(
8  "com.typesafe.akka" %% "akka-actor" % "2.5.19",
9  "com.typesafe.akka" %% "akka-cluster-tools" % "2.5.19",
10  "com.typesafe.akka" %% "akka-distributed-data" % "2.5.19"
11)

resources/application.conf

1akka {
2   actor.provider = "cluster"
3   remote {
4     netty.tcp.port = 0
5     netty.tcp.hostname = "localhost"
6   }
7
8  extensions = ["akka.cluster.ddata.DistributedData"]
9
10  cluster {
11    seed-nodes = [
12      "akka.tcp://DDataSystem@localhost:2551",
13      "akka.tcp://DDataSystem@localhost:2552"]
14
15    # auto downing is NOT safe for production deployments.
16    # you may want to use it during development, read more about it in the docs.
17    #
18    # auto-down-unreachable-after = 10s
19  }
20
21}

DDataUpdator.scala

1import akka.actor._
2import akka.cluster.ddata._
3import Replicator._
4import akka.cluster.Cluster
5import com.typesafe.config.ConfigFactory
6
7import scala.concurrent.duration._
8
9object DDataUpdator {
10
11  case object IncCounter
12  case class AddToSet(item: String)
13  case class AddToMap(item: String)
14  case object ReadSet
15  case object ReadMap
16  case object ShutDownDData
17
18
19  val KeyCounter = GCounterKey("counter")
20  val KeySet = ORSetKey[String]("gset")
21  val KeyMap = ORMultiMapKey[Long, String]("ormap")
22
23  val timeout = 300 millis
24  val writeAll = WriteAll(timeout)
25  val readAll = ReadAll(timeout)
26
27
28  def create(port: Int): ActorRef = {
29    val config = ConfigFactory.parseString(s"akka.remote.netty.tcp.port = $port")
30      .withFallback(ConfigFactory.load())
31    val system = ActorSystem("DDataSystem",config)
32    system.actorOf(Props[DDataUpdator],s"updator-$port")
33  }
34
35}
36
37class DDataUpdator extends Actor with ActorLogging {
38  import DDataUpdator._
39  implicit val cluster = Cluster(context.system)
40  val replicator = DistributedData(context.system).replicator
41
42
43  replicator ! Subscribe(KeyCounter,self)
44  replicator ! Subscribe(KeySet,self)
45  replicator ! Subscribe(KeyMap,self)
46
47  override def receive: Receive = {
48    case IncCounter =>
49       log.info(s"******* Incrementing counter... *****")
50       replicator ! Update(KeyCounter,GCounter(),writeAll)(_ + 1)
51    case UpdateSuccess(KeyCounter,_) =>
52      log.info(s"********** Counter updated successfully ********")
53    case UpdateTimeout(KeyCounter,_) =>
54      log.info(s"******* Counter update timed out! *****")
55    case ModifyFailure(KeyCounter,msg,err,_) =>
56      log.info(s"******* Counter update failed with error: ${msg} *****")
57    case StoreFailure(KeyCounter,_) =>
58      log.info(s"******* Counter value store failed! *****")
59    case c @ Changed(KeyCounter) ⇒
60      val data = c.get(KeyCounter)
61      log.info("********Current count: {}*******", data.getValue)
62
63
64    case AddToSet(item) =>
65      replicator ! Update(KeySet,ORSet.empty[String],writeAll)(_ + item)
66    case UpdateSuccess(KeySet,_) =>
67      log.info(s"**********Add to ORSet successfully ********")
68    case UpdateTimeout(KeySet,_) =>
69      log.info(s"******* Add to ORSet timed out! *****")
70    case ModifyFailure(KeySet,msg,err,_) =>
71      log.info(s"******* Add to ORSet failed with error: ${msg} *****")
72    case StoreFailure(KeySet,_) =>
73      log.info(s"******* ORSet items store failed! *****")
74    case c @ Changed(KeySet) =>
75      val data = c.get(KeySet)
76      log.info("********Items in ORSet: {}*******", data.elements)
77    case ReadSet =>
78      replicator ! Get(KeySet,readAll)
79    case g @ GetSuccess(KeySet, req) =>
80      val value = g.get(KeySet)
81      log.info("********Current items read in ORSet: {}*******", value.elements)
82    case NotFound(KeySet, req) =>
83      log.info("******No item found in ORSet!!!*******")
84
85
86
87    case AddToMap(item) =>
88       replicator ! Get(KeyCounter,readAll,Some(AddToMap(item)))
89    case g @ GetSuccess(KeyCounter,Some(AddToMap(item))) =>
90      val idx: Long = g.get(KeyCounter).getValue.longValue()
91      log.info(s"*********** got counter=${idx} with item: $item ************")
92      replicator ! Update(KeyMap,ORMultiMap.empty[Long,String],writeAll)(_ + (idx -> Set(item)))
93      replicator ! Update(KeyCounter,GCounter(),writeAll)(_ + 1)
94    case c @ Changed(KeyMap) =>
95      val data = c.get(KeyMap).entries
96      log.info("******** Items in ORMultiMap: {}*******", data)
97    case ReadMap =>
98      replicator ! Get(KeyMap,readAll)
99    case g @ GetSuccess(KeyMap, req) =>
100      val value = g.get(KeyMap)
101      log.info("********Current items read in ORMultiMap: {}*******", value.entries)
102    case NotFound(KeyMap, req) =>
103      log.info("****** No item found in ORMultiMap!!! *******")
104
105
106
107    case ShutDownDData => context.system.terminate()
108
109  }
110}
111
112
113object DDataDemo extends App {
114  import DDataUpdator._
115
116  val ud1 = create(2551)
117  val ud2 = create(2552)
118  val ud3 = create(2553)
119  scala.io.StdIn.readLine()
120
121  ud1 ! IncCounter
122  ud2 ! AddToSet("Apple")
123  ud1 ! AddToSet("Orange")
124
125  scala.io.StdIn.readLine()
126
127  ud2 ! IncCounter
128  ud2 ! AddToSet("Pineapple")
129  ud1 ! IncCounter
130  ud1 ! AddToMap("Cat")
131
132  scala.io.StdIn.readLine()
133
134  ud1 ! AddToMap("Dog")
135  ud2 ! AddToMap("Tiger")
136  scala.io.StdIn.readLine()
137
138  ud3 ! ReadSet
139  ud3 ! ReadMap
140  scala.io.StdIn.readLine()
141
142
143   ud1 ! ShutDownDData
144   ud2 ! ShutDownDData
145   ud3 ! ShutDownDData
146}