AKKA框架持久化入门样例

背景

我们在开发的过程中可能会在内存中操作数据，但是可能会遇到突然服务器断电、网线被挖等情况。这就需要将内存中的数据持久化，在程序重启的时候依然能够恢复。

AKKA介绍

Akka is a toolkit for building highly concurrent, distributed, and resilient message-driven applications for Java and Scala

Akka是一个用于为Java和Scala构建高度并发，分布式和弹性的消息驱动应用程序的工具包

虽然AKKA不是专门用来持久化的，但是其中的持久化部分可以拿出来使用。

本文参照AKKA官方文档，展示了AKKA持久化的一个样例。

第一步：配置JDK和Maven环境

这里推荐JDK1.8和Maven3

第二步：创建Maven工程

本人习惯采用IDEA

如果创建完工程有提示

第三步：添加依赖

修改pom.xml文件，其实只有两个依赖就够用了。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>TestAkka</groupId><artifactId>TestAkka</artifactId><version>1.0-SNAPSHOT</version><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>8</source><target>8</target></configuration></plugin></plugins></build><dependencies><dependency><groupId>com.typesafe.akka</groupId><artifactId>akka-actor_2.12</artifactId><version>2.5.22</version></dependency><dependency><groupId>org.iq80.leveldb</groupId><artifactId>leveldb</artifactId><version>0.7</version></dependency></dependencies></project>

第四步：编写代码

总共需要四个文件

（1）MyData 这个类是用来封装数据的，也就是说，我们希望进行持久化的数据可以封装到这个里面。需要注意的是，这个类必须要实现Serializable接口

import java.io.Serializable;public class MyData implements Serializable {private static final long serialVersionUID = 1L;private String name;public MyData(String name){this.name= name;}@Overridepublic String toString() {return "MyData{" +"name='" + name + '\'' +'}';}
}

（2）Actor类

这个类要继承AbstractPersistentActor类，并能覆盖其中三个方法。

import akka.persistence.AbstractPersistentActor;
import akka.persistence.SnapshotOffer;import java.util.ArrayList;
import java.util.List;public class MyPersistentActor extends AbstractPersistentActor {private List<MyData> state = new ArrayList<MyData>();@Overridepublic String persistenceId() {return "Kangyucheng";}@Overridepublic Receive createReceiveRecover() {return receiveBuilder().match(MyData.class, e -> state.add(e)).match(SnapshotOffer.class, ss -> state = (List<MyData>) ss.snapshot()).build();}@Overridepublic Receive createReceive() {return receiveBuilder().match(MyData.class, c -> {persist(c, event -> state.add(event));}).matchEquals("snap", s -> saveSnapshot(new ArrayList<>(state))).matchEquals("print", s -> System.out.println(state)).build();}}

（3）Main测试类

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;public class Main {public static void main(String... args) throws Exception {//1.创建 persistentActor,注意MyPersistentActor对应的是我们自己创建的Actorfinal ActorSystem system = ActorSystem.create();final ActorRef persistentActor = system.actorOf(Props.create(MyPersistentActor.class));//2.调用tell方法String random = String.valueOf((int)(Math.random()*100));persistentActor.tell(new MyData("kangyucheng"+random), null);persistentActor.tell(new MyData("kyc"), null);persistentActor.tell("snap", null);persistentActor.tell("print", null);//3.由于系统是异步持久化，所以要等一下Thread.sleep(10000);system.terminate();}
}

（4） AKKA配置类application.conf这个文件放在了resources文件下面

主要是配置持久化文件的存储位置

akka.persistence.journal.plugin = "akka.persistence.journal.leveldb"
akka.persistence.snapshot-store.plugin = "akka.persistence.snapshot-store.local"akka.persistence.journal.leveldb.dir = "target/example/journal"
akka.persistence.snapshot-store.local.dir = "target/example/snapshots"# DO NOT USE THIS IN PRODUCTION !!!
akka.persistence.journal.leveldb.native = false

第五步：运行程序

（1）第一次运行

发现target文件夹下出现了配置文件中的目录，这个目录存储的就是持久化的文件

控制台打出来了List

（2）第二次运行

可以看见第一次运行存储在内存中的

（3）第三次运行

可见前两次运行的结果