Akka2使用探索7——「雲計算」示例（Actor、Future、Remoting、Router、Deploy、異步、併發使用Demo）

假設有一個很耗時的運算，單臺機器已經無法知足需求，這時你能夠想到由多臺計算機協做完成。具體怎麼作呢。

舉個很簡單的例子，假設這個耗時的運算是從1加到100000，你如今有兩臺服務器，能夠讓這兩臺服務器分別完成從1加到50000，和從50001加到100000，而後本機完成這兩個結果之和。

 

兩臺服務器分別啓動兩個akka Server，同時還有一個CalcActor。這個計算actor接收兩個參數：Integer start和Integer end，能夠從start一直加到end，最後將結果返回給發送者：getSender().tell(result)。

@Log4j
class CalcActor extends UntypedActor {
    @Override
    void onReceive(Object message) {
        log.debug "CalcActor received: ${message}----self:${getSelf()},sender:${getSender()}"
        if (message instanceof String) {
            String[] args = message.split(",")
            int start = Integer.parseInt(args[0])
            int end = Integer.parseInt(args[1])
            double result = 0d
            println("start calc:" + start + " upto " + end)
            start.upto(end) {
                result += it
            }
            sleep(5000) //模擬還要額外耗時5秒
            println("result:" + result)
            getSender().tell(result)
        } else {
            unhandled(message)
        }
    }
}
兩個服務器分別爲：

AkkaServerApp serverA = new AkkaServerApp("sc", "10.68.3.122", 8888, "calc") //AkkaSystemName爲sc，ip爲10.68.3.122，端口爲8888，serviceName爲calc。

    AkkaServerApp serverA = new AkkaServerApp("sp", "10.68.3.124", 8888, "calc")//AkkaSystemName爲sp，ip爲10.68.3.124，端口爲8888，serviceName爲calc。

主要的代碼在客戶端：

public static void main(String[] args) throws Exception {

        final AkkaServerApp app = new AkkaServerApp("xwc", "127.0.0.1", 6666, "client");//客戶端akka配置
        ActorRef remoteCalcA1 = app.getSystem().actorOf(new Props(CalcActor.class)..withDeploy(new Deploy(new RemoteScope(new Address("akka", "sc", "10.68.3.122", 8888)))), "clientCalcA1");//將CalcActor發佈到遠程10.68.3.122上
        ActorRef remoteCalcA2 = app.getSystem().actorOf(new Props(CalcActor.class)..withDeploy(new Deploy(new RemoteScope(new Address("akka", "sc", "10.68.3.122", 8888)))), "clientCalcA2");//將CalcActor發佈到遠程10.68.3.124上

        final List<Future<Double>> frs = new ArrayList<Future<Double>>();//異步返回結果Future存放在list中
//tell只請求，是否響應它徹底不知道。ask是請求，並明確知道將來會相應。        
// remoteCalcA.tell("1,10000", app.getServerActor());
// remoteCalcB.tell("10001,20000", app.getServerActor());
        Future f1 = akka.pattern.Patterns.ask(remoteCalcA1, "1,50000", 150000);//讓遠程122計算從1加到50000，超時時間爲150秒
        Future f2 = akka.pattern.Patterns.ask(remoteCalcA1, "50001,100000", 150000);//併發地讓遠程124計算從50001加到100000，超時時間150秒
        frs.add(f1);
        frs.add(f2);

        Future<Iterable<Double>> future = Futures.sequence(frs, app.getSystem().dispatcher());將將來返回的結果轉換成Future<Iterable<Double>>
        Future<Double> fr = future.map(new Mapper<Iterable<Double>, Double>() {

            @Override
            public Double apply(Iterable<Double> parameter) {
                Double result = 0d;
                for (Double s : parameter) {//計算兩個服務器返回的結果
                    result += s;
                }
                return result;
            }

        });

        fr.onSuccess(new OnSuccess<Double>() {
            @Override
            public void onSuccess(Double result) {
                System.out.println("雲計算返回結果-----" + result);
            }
        });
    }

 

還能夠讓服務器併發處理：把給從1加到50000的任務分紅5個線程並行處理：1..10000,10001..20000,20001..30000,30001..40000,40001..50000，這樣能更好地提升效率。

若是按上面的方法僅僅是發佈多個remote actor：

ActorRef remoteCalcAn = app.getSystem().actorOf(new Props(CalcActor.class)..withDeploy(new Deploy(new RemoteScope(new Address("akka", "sc", "10.68.3.122", 8888)))), "clientCalcAn");

是無法提升效率的，由於這時的CalcActor是單線程的，它只會先接收1..10000，處理完後再接收10001..20000並處理。。。。。

使其可以並行處理很簡單，建立remoteActor時加上withRoute便可：

ActorRef remoteCalcAn = app.getSystem().actorOf(new Props(CalcActor.class).withRouter(new RoundRobinRouter(5)).withDeploy(new Deploy(new RemoteScope(new Address("akka", "sc", "10.68.3.122", 8888)))), "clientCalcAn");  //RoundRobinRouter的參數5能夠理解爲分配5個線程並行處理

代碼跟上面基本相同

public static void main(String[] args) throws Exception {

        final AkkaServerApp app = new AkkaServerApp("xwc", "127.0.0.1", 6666, "client");
        ActorRef remoteCalcA1 = app.getSystem().actorOf(new Props(CalcActor.class).withRouter(new RoundRobinRouter(4)).withDeploy(new Deploy(new RemoteScope(new Address("akka", "sc", "10.68.3.122", 8888)))), "clientCalcA1");    
        ActorRef remoteCalcB1 = app.getSystem().actorOf(new Props(CalcActor.class).withRouter(new RoundRobinRouter(4)).withDeploy(new Deploy(new RemoteScope(new Address("akka", "sp", "10.68.3.124", 8888)))), "clientCalcB1");
   
        final List<Future<Double>> frs = new ArrayList<Future<Double>>();
        
        Future f1 = akka.pattern.Patterns.ask(remoteCalcA1, "1,10000", 150000);
        Future f2 = akka.pattern.Patterns.ask(remoteCalcA1, "10001,20000", 150000);
        Future f3 = akka.pattern.Patterns.ask(remoteCalcA1, "20001,30000", 150000);
        Future f4 = akka.pattern.Patterns.ask(remoteCalcA1, "30001,40000", 150000);
        Future f5 = akka.pattern.Patterns.ask(remoteCalcB1, "40001,50000", 150000);
        Future f6 = akka.pattern.Patterns.ask(remoteCalcB1, "50001,60000", 150000);
        Future f7 = akka.pattern.Patterns.ask(remoteCalcB1, "60001,70000", 150000);
        Future f8 = akka.pattern.Patterns.ask(remoteCalcB1, "70001,80000", 150000);
        frs.add(f1);
        frs.add(f2);
        frs.add(f3);
        frs.add(f4);
        frs.add(f5);
        frs.add(f6);
        frs.add(f7);
        frs.add(f8);

        Future<Iterable<Double>> future = Futures.sequence(frs, app.getSystem().dispatcher());
        Future<Double> fr = future.map(new Mapper<Iterable<Double>, Double>() {

            @Override
            public Double apply(Iterable<Double> parameter) {
                Double result = 0d;
                for (Double s : parameter) {
                    result += s;
                }
                return result;
            }

        });

        fr.onSuccess(new OnSuccess<Double>() {
            @Override
            public void onSuccess(Double result) {
                System.out.println("雲計算返回從1加到80000的結果-----" + result);
            }
        });
    }