Java并发 -- Actor模型

public class HelloActor extends UntypedAbstractActor {
    // 该Actor在收到消息message后，会打印Hello message
    @Override
    public void onReceive(Object message) throws Throwable {
        System.out.printf("Hello %s%n", message);
    }

    public static void main(String[] args) {
        // 创建Actor系统，Actor不能脱离ActorSystem存在
        ActorSystem system = ActorSystem.create("HelloSystem");
        // 创建HelloActor
        ActorRef actorRef = system.actorOf(Props.create(HelloActor.class));
        // 发送消息给HelloActor
        actorRef.tell("Actor", ActorRef.noSender());
    }
}

public class CounterActor extends UntypedAbstractActor {
    private int counter = 0;

    @Override
    public void onReceive(Object message) throws Throwable {
        if (message instanceof Number) {
            counter += ((Number) message).intValue();
        } else {
            System.out.println(counter);
        }
    }

    // 没有锁，也没有CAS，但程序是线程安全的
    public static void main(String[] args) throws InterruptedException {
        ActorSystem system = ActorSystem.create("CounterSystem");
        ExecutorService pool = Executors.newFixedThreadPool(4);
        ActorRef actorRef = system.actorOf(Props.create(CounterActor.class));
        // 生成4*100_000个消息
        for (int i = 0; i < 4; i++) {
            pool.execute(() -> {
                for (int j = 0; j < 100_000; j++) {
                    actorRef.tell(1, ActorRef.noSender());
                }
            });
        }
        pool.shutdown();
        // 等待actorRef处理完所有消息
        TimeUnit.SECONDS.sleep(5);
        // 打印结果
        actorRef.tell("", ActorRef.noSender()); // 400_000
    }
}