使用Akka持久化——持久化与快照_system.eventstream().publish

前言

对于Java web而言，一个用户的HTTP请求最终会转换为一条Java线程处理。HTTP本身是无状态的，具体的请求逻辑一般也是无状态的。如果进程奔溃或者系统宕机，用户会发觉当前网页不可用之类的错误。虽然会影响一些用户体验，但是只要服务重启了，用户依然可以完成他的请求并满足其需要。但是有些情况下则势必会造成混乱甚至恐慌，例如跨行转账。用户从自己A银行的账户转账1万元至自己在B银行的账户，如果转出的动作成功了，但是转入却失败了，用户的心情是可想而知的，自己的财产不翼而飞了！一种解决的方式是引入事务，在此场景下还必须是分布式事务。如果只是银行内部实现分布式事务多少还是可行的，但是不同银行之间要实现的成本是可想而知的，甚至不可行的。如果A银行转出时对用户的状态作持久化，B银行对收到的转入请求也进行持久化，那么恢复用户的损失才有可能。

以上啰里啰嗦说了这么多，无非就是抛出个引子，进而介绍Akka提供的持久化功能。

Akka的持久化架构

• UntypedPersistentActor: 是一个持久化、有状态的Actor。它能够将将事件持久化到日志并且以线程安全的方式重新执行这些事件。它能够被用于实现命令和事件源的Actor。当一个集成了UntypedPersistentActor的Actor启动或者重启时，日志化的消息被重新发送给此Actor，以便于通过这些消息回复内部状态。
• UntypedPersistentView: 视图是一个持久化、有状态的Actor，它接收由另一个持久化Actor写入日志的消息。视图本身不会日志化新的消息，它仅仅从一个持久化Actor的复制消息流来更新内部状态。
• UntypedPersistentActorAtLeastOnceDelivery: 用于实现向目标至少发送一次消息的语义，也适用于发送者和接收者的JVM进程奔溃。
• AsyncWriteJournal: 异步存储发送给持久化Actor的消息序列的日志。应用程序能够通过持久化Actor控制哪个消息被日志化，哪个消息不用被日志化。日志为每条消息维护一个不断增加的序列号。日志存储的底层实现是可插拔的。Akka的持久化扩展自带一个叫做"leveldb"，向本地文件系统写入的日志插件。Akka社区里还有更多日志存储插件提供。
• Snapshot store: 快照存储对持久化Actor或持久化视图的内部状态的快照进行持久化。快照用于优化回复的时间。快照存储的底层是可插拔的。Akka持久化扩展自带一个向本地文件系统写入的“本地”快照存储插件。Akka社区里还有更多快照存储插件提供。

本文基于Akka官网提供的持久化例子，并对其进行一些适应性改造，将着重介绍UntypedPersistentActor、AsyncWriteJournal及Snapshot store的应用。

配置

有关Akka的日志持久化和快照持久化的配置如下：

  persistence {
    journal {
      plugin = "akka.persistence.journal.leveldb"
      leveldb.dir = "target/example/journal"
      leveldb.native = false
    }
    snapshot-store {
      plugin = "akka.persistence.snapshot-store.local"
      local.dir = "target/example/snapshots"
    }
  }

根据配置，我们知道日志插件使用了leveldb，leveldb的存储目录为当前项目编译路径下的example/journal路径下。快照插件使用了local，存储路径与前者相同。

持久化Actor的例子

消息与状态

本例子中需要用到Cmd和Evt两种消息，Cmd代表命令，Evt代表事件。ExampleState代表我们例子中的状态。以上三个类的定义如下：

public interface Persistence {
 
	public static class Cmd implements Serializable {
		private static final long serialVersionUID = 1L;
		private final String data;
 
		public Cmd(String data) {
			this.data = data;
		}
 
		public String getData() {
			return data;
		}
	}
 
	public static class Evt implements Serializable {
		private static final long serialVersionUID = 1L;
		private final String data;
 
		public Evt(String data) {
			this.data = data;
		}
 
		public String getData() {
			return data;
		}
	}
 
	public static class ExampleState implements Serializable {
		private static final long serialVersionUID = 1L;
		private final ArrayList<String> events;
 
		public ExampleState() {
			this(new ArrayList<String>());
		}
 
		public ExampleState(ArrayList<String> events) {
			this.events = events;
		}
 
		public ExampleState copy() {
			return new ExampleState(new ArrayList<String>(events));
		}
 
		public void update(Evt evt) {
			events.add(evt.getData());
		}
 
		public int size() {
			return events.size();
		}
 
		@Override
		public String toString() {
			return events.toString();
		}
	}
}

上面代码展示的Cmd和Evt都很简单，它们有一样的data字段作为内容。ExampleState中维护了一个列表，次列表用于缓存所有的事件内容。

持久化Actor的实现

在具体介绍本例中持久化Actor之前，先看看其实现，其代码清单如下：