AKKA框架持久化入门样例_akka 持久化db

背景

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

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;
    }
 
    @Override
    public 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>();
 
    @Override
    public String persistenceId() {
        return "Kangyucheng";
    }
    @Override
    public Receive createReceiveRecover() {
        return receiveBuilder()
                .match(MyData.class, e -> state.add(e))
                .match(SnapshotOffer.class, ss -> state = (List<MyData>) ss.snapshot())
                .build();
    }
 
    @Override
    public 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对应的是我们自己创建的Actor
        final 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）第三次运行

可见前两次运行的结果