akka学习教程(十) agent

迈不过友情╰ 2022-07-11 04:26 187阅读 0赞

# akka系列文章目录 #

*  [akka学习教程(十四) akka分布式实战][akka_ akka]
 *  [akka学习教程(十三) akka分布式][akka_ akka 1]
 *  [akka学习教程(十二) Spring与Akka的集成][akka_ Spring_Akka]
 *  [akka学习教程(十一) akka持久化][akka_ akka 2]
 *  [akka学习教程(十) agent][akka_ agent]
 *  [akka学习教程(九) STM软件事务内存][akka_ STM]
 *  [akka学习教程(八) Actor中的Future-询问模式][akka_ Actor_Future-]
 *  [akka学习教程(七) 内置状态转换Procedure][akka_ _Procedure]
 *  [akka学习教程(六) 路由器Router][akka_ _Router]
 *  [akka学习教程(五) inbox消息收件箱][akka_ inbox]
 *  [akka学习教程(四) actor生命周期][akka_ actor]
 *  [akka学习教程(三) 不可变对象][akka_]
 *  [akka学习教程(二)HelloWord][akka_HelloWord]
 *  [akka学习教程(一)简介][akka]

我们知道在JDK中，编写多线程代码时要谨慎处理临界区的数据，可以加锁或者使用JDK自带的CAS库：atomic相关包。

那么在akka中怎么处理呢？akka给我们提供了一个agent。可以使用agent来实现共享变量的安全处理。

下面示例为10个actor并发累加`countAgent`变量。每个累加10000次。如果正确的话，最终结果应该是10W。

package akka;
    
    import akka.actor.*;
    import akka.agent.Agent;
    import akka.dispatch.ExecutionContexts;
    import akka.dispatch.Futures;
    import akka.dispatch.Mapper;
    import akka.dispatch.OnComplete;
    import akka.event.Logging;
    import akka.event.LoggingAdapter;
    import com.typesafe.config.ConfigFactory;
    import scala.concurrent.Future;
    import scala.concurrent.duration.Duration;
    
    import java.util.concurrent.ConcurrentLinkedQueue;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.TimeoutException;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /** * Created by liubenlong on 2017/1/12. */
    public class AgentTest extends UntypedActor { 
    
        private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    
        @Override
        public void onReceive(Object o) throws Throwable {
            if(o instanceof Integer){
                for(int i = 0 ; i < 10000 ; i ++){
                    Future<Integer> future = countAgent.alter(new Mapper<Integer, Integer>() {
                        @Override
                        public Integer apply(Integer parameter) {
                            return parameter + 1;
                        }
                    });
                    queue.add(future);
                }
    
                getContext().stop(getSelf());//完成任务，关闭自己
            }else{
                unhandled(o);
            }
        }
    
    
        public static CountDownLatch latch = new CountDownLatch(10);
        public static Agent<Integer> countAgent = Agent.create(0, ExecutionContexts.global());
        public static ConcurrentLinkedQueue<Future<Integer>> queue = new ConcurrentLinkedQueue<>();
    
        public static void main(String [] args) throws InterruptedException, TimeoutException {
            ActorSystem system = ActorSystem.create("inbox", ConfigFactory.load("akka.conf"));
    
            ActorRef[] actorRefs = new ActorRef[10];
            for(int i = 0 ; i < 10 ; i ++){
                actorRefs[i] = system.actorOf(Props.create(AgentTest.class), "AgentTest" + i);
            }
    
            Inbox inbox = Inbox.create(system);
            for (ActorRef ref : actorRefs) {
                inbox.send(ref, 1);
                inbox.watch(ref);
            }
    
            System.out.println("countAgent 1:" + countAgent.get());
    
    
            /* * 这里本来想用JDK里的AtomicInteger, 但是不行，不适用于这里的actor */
    
    
            //等待所有actor执行完毕
            int closeCount = 0;
            while(true){
                Object o = inbox.receive(Duration.create(1, TimeUnit.SECONDS));
                if(o instanceof Terminated){
                    closeCount ++;
                    if(closeCount == actorRefs.length){
                        break;
                    }
                }else{
                    System.out.println("o:" + o);
                }
            }
    
            System.out.println("countAgent 2:" + countAgent.get());
    
            //等待所有累加线程完成
            Futures.sequence(queue, system.dispatcher()).onComplete(new OnComplete<Iterable<Integer>>() {
                @Override
                public void onComplete(Throwable throwable, Iterable<Integer> integers) throws Throwable {
                    System.out.println("countAgent 3:" + countAgent.get());
                    system.shutdown();
                }
            }, system.dispatcher());
    
        }
    }

运行结果：

countAgent 1:0
    countAgent 2:10
    countAgent 3:100000

# 参考资料 #

*  书籍《java高并发程序设计》
 *  [AKKA官方文档][AKKA]

[akka_ akka]: http://blog.csdn.net/liubenlong007/article/details/54603296
[akka_ akka 1]: http://blog.csdn.net/liubenlong007/article/details/54601998
[akka_ Spring_Akka]: http://blog.csdn.net/liubenlong007/article/details/54601448
[akka_ akka 2]: http://blog.csdn.net/liubenlong007/article/details/54584303
[akka_ agent]: http://blog.csdn.net/liubenlong007/article/details/54576611
[akka_ STM]: http://blog.csdn.net/liubenlong007/article/details/54576412
[akka_ Actor_Future-]: http://blog.csdn.net/liubenlong007/article/details/54575312
[akka_ _Procedure]: http://blog.csdn.net/liubenlong007/article/details/54574878
[akka_ _Router]: http://blog.csdn.net/liubenlong007/article/details/54574064
[akka_ inbox]: http://blog.csdn.net/liubenlong007/article/details/54377820
[akka_ actor]: http://blog.csdn.net/liubenlong007/article/details/54093889
[akka_]: http://blog.csdn.net/liubenlong007/article/details/54093794
[akka_HelloWord]: http://blog.csdn.net/liubenlong007/article/details/53782971
[akka]: http://blog.csdn.net/liubenlong007/article/details/53782966
[AKKA]: http://akka.io/docs/