akka 编程入门 - 状态机

发表于2019-11-30，长度3891， 193个单词， 11分钟读完

由于akka是基于事件驱动的actor模型，所以很容易实现状态机。这里用官方的一个例子演示一下。不过为了更加清晰，我将大部分结构改成了中文名。代码可以从https://gitee.com/somefuture/akka-java-101/tree/master/manxi-fsm下载。

状态机的逻辑是：初始为闲置的，可以为其设置目标actor，然后往其中的队列添加对象。当对象集合不为空时状态变更为活动的，此时可以令其将对象全部发给目标actor。发送以后状态机回到闲置状态。触发给目标actor发送信息的动机有两个：一个是发送清空命令，一个是队列1秒内没有再接到添加对象的请求引起超时。

先创建测试流程：

class Start extends AbstractActor {
    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .matchAny(m -> {
                    final ActorRef buncher = getContext().system().actorOf(Props.create(Buncher.class));
                    final ActorRef probe = getContext().system().actorOf(Props.create(Target.class));

                    buncher.tell(new 设置目标事件(probe), getSelf());
                    buncher.tell(new 入队事件(42), probe);
                    buncher.tell(new 入队事件(43), probe);
                    buncher.tell(new 入队事件(44), probe);
                    buncher.tell(清空事件.Flush, probe);
                    buncher.tell(new 入队事件(45), probe);
                })
                .build();
    }
}

其中buncher是状态机，probe是目标actor。

之后我们向状态机发送了一系列请求：先是设置了目标为probe，然后连续三次请求入队，接下来是清空事件，会触发向目标发送消息。然后又请求了一次入队，后面没有新请求，会触发超时事件。

下面看一下状态机的实现：

import static info.manxi.状态枚举.活动的;
import static info.manxi.状态枚举.闲置;
import static info.manxi.初始化数据.初始数据;
class Buncher extends AbstractFSM<状态枚举, 可接受数据> {
    {
        startWith(闲置, 初始数据);

        when(
                闲置,
                matchEvent(
                        设置目标事件.class,
                        初始化数据.class,
                        (setTarget, uninitialized) ->
                                stay().using(new 运行时数据(setTarget.getRef(), new LinkedList<>()))));

        onTransition(
                matchState(
                        活动的,
                        闲置,
                        () -> {
                            final UnitMatch<可接受数据> m =
                                    UnitMatch.create(
                                            matchData(
                                                    运行时数据.class,
                                                    todo ->
                                                            todo.getTarget().tell(new Batch(todo.getQueue()), getSelf())));
                            m.match(stateData());
                        })
                        .state(
                                闲置,
                                活动的,
                                () -> {
                                    /* 其他逻辑 */
                                }));

        when(
                活动的,
                Duration.ofSeconds(1L),
                matchEvent(
                        Arrays.asList(清空事件.class, StateTimeout()),
                        运行时数据.class,
                        (event, todo) -> goTo(闲置).using(todo.copy(new LinkedList<>()))));

        whenUnhandled(
                matchEvent(
                        入队事件.class,
                        运行时数据.class,
                        (queue, todo) -> goTo(活动的).using(todo.addElement(queue.getObj())))
                        .anyEvent(
                                (event, state) -> {
                                    log().warning("未处理的事件 {} 对于状态 {}/{}", event, stateName(), state);
                                    return stay();
                                }));

        initialize();
    }
}

主要有这么几个过程：

状态机也是actor，但是继承的是akka.actor.AbstractFSM。
使用startWith方法设置状态机初始状态和数据
通过when方法匹配状态机在某个状态下接收到某种事件的行为，其他场景都使用whenUnhandled匹配
在onTransition中通过matchState定义状态转移时的行为
通过initialize方法启动状态机

下面进行源码解析：

startWith(闲置, 初始数据);表示状态机初始为闲置状态，其中的数据是“初始数据”
第一个when表示在闲置状态下如果设置目标actor，则维持状态不变（stay()方法），using表示当前使用什么数据
第二个when表示在活动状态下如果收到清空请求或超市事件（1秒超时），则goto到闲置状态，目标不变，数据置空
whenUnhandled处理的是入队请求（当然可以写到一个新的when里面），收到入队对象就goto到活动的状态（如果本来就是活动的就不变），并将对象入队。对于其他事件则打印log并维持状态。
从活动的状态转为闲置时，会给目标actor发送信息，封装在Batch里。

这样我们的测试流程就很清晰了。输出如下：