用Java实现Actor模型（模仿Skynet）

Actor模型是一种常见的并发模型，与最常见的并发模型——共享内存（同步锁）不同，它将程序分为许多独立的计算单元——Actor，每个Actor独立管理自己的资源，不同Actor之间通过消息传递来交互。它的好处是全异步执行，不会造成线程阻塞，从而提升CPU使用率，另外由于线程之间是异步交互，所以也不用考虑加锁和线程同步的问题。

Actor模型在业界有许多应用，例如游戏服务器框架Skynet、编程语言Erlang。

因为历史原因，Java下的Actor模型应用较少，知名的只有基于Scala的Akka。而且Actor模型也不是万能的，异步编程会需要编写更多的回调代码，原本的一步需要拆分成若干步来处理，无疑增加了代码编写复杂度（callback hell）。

本文以学习和研究为目的，使用Java实现一个简单Actor模型，功能上模仿Skynet，支持的功能包括：

1. Actor基础功能：消息发送接收、异步处理等。

2. 集群功能：支持多节点之间通信。

3. 非阻塞的sleep和网络通信。

完整的源代码在可以在Github获取。以下是部分关键代码以及设计思路讲解。

Actor

Actor是Actor模型中的核心概念，每个Actor独立管理自己的资源，与其他Actor之间通信通过Message。

这里的每个Actor由单线程驱动，相当于Skynet中的服务。Actor不断从mailbox中获取尚未处理的Message，mailbox使用的结构是无界阻塞的LinkedBlockingQueue。

Actor类是抽象类，其中处理消息的handleMessage方法为抽象方法，需要每个具体类来重载实现。

public abstract class Actor {

    private Node node;

    private String name;

    private final BlockingQueue<Message> mailbox = new LinkedBlockingQueue<>();

    private Thread actorThread;

    public Node getNode() {
        return node;
    }

    public void setNode(Node node) {
        this.node = node;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }

    public void start() {
        actorThread = new Thread(() -> {
            ActorSystem.setThreadLocalActor(this);
            for(;;) {
                try {
                    Message message = mailbox.take();
                    try {
                        handleMessage(message);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                } catch (InterruptedException ignore) {
                    // ignore                }
            }
        });

        actorThread.start();
    }

    public void act(Message msg) {
        mailbox.offer(msg);
    }

    protected abstract void handleMessage(Message message);}

Node

Node代表节点，与Skynet中节点意义相同。它是一个独立的Java进程，有自己的IP和端口，Node之间通过异步的网络通信发送和接收消息。一个Node中可以运行多个Actor，一个Actor仅可与一个Node绑定。

Node的唯一标识也是它的name，与Actor的name稍有不同，Node的name是全局唯一，而Actor的name是Node内唯一。

public abstract class Node {

    /**     * 名字     * 需要是唯一的，按名字查找     */
    private String name;

    private InetSocketAddress address;

    public String getName() {
        return name;
    }

    public void setName(String nodeName) {
        name = nodeName;
    }

    public void setAddress(InetSocketAddress address) {
        this.address = address;
    }}

ActorSystem

ActorSystem是Actor的管理系统，也是外部调用API的主要入口，提供本框架中的主要功能：创建Actor、发送消息、休眠Actor、网络通信等。下面分别详细说明。

ActorSystem初始化

分为以下三步：

首先是调用conf方法读取集群配置，包括每个Node的name和address。

其次是调用bindNode方法绑定当前Node。

最后是调用start方法初始化自身，包括对定时器的初始化和Netty服务端的初始化。之所以引入定时器，是因为无阻塞sleep需要用到，这个具体后面再说，另外也可以用于扩展实现通用的定时任务功能。Node之间发送消息都是异步的，客户端和服务端都使用了Netty做异步网络通信。

public class ActorSystem {

    private static Map<String, InetSocketAddress> clusterConfig;

    /**     * 当前绑定到的节点     */
    private static Node currNode;

    private final static Map<String, Actor> actors = new HashMap<>();

    /**     * 维护线程与Actor的对应关系     */
    private final static ThreadLocal<Actor> currThreadActor = new ThreadLocal<>();

    /**     * 客户端Netty bootstrap     */
    private static Bootstrap clientBootstrap;

    /**     * 维护节点与通道的对应关系     */
    private final static Map<String, Channel> channels = new ConcurrentHashMap<>();

    private static void startNettyBootstrap() {
        try {
            // 先启动服务端bootstrap            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)))
                     .addLast(new ObjectEncoder(