初識Akka之Router

背景介紹

簡單來說Akka是一個用于建構服務端分布式應用的架構，其具有高并發、可擴充、容錯性等特點。Akka中的基本機關是Actor，Actor可以與線程來作為類比，有自己的辨別符、存儲空間、排程政策、生命周期等。每個Actor互相之間的互動都是通過消息(可以是字元串、二進制資料、自定義對象等)來實作的，這樣就避免了每個Actor的耦合。更重要的是生成一個Actor隻占用40多個位元組的存儲空間(我自己測試的是50位元組左右)，1G記憶體的情況下，可以生成250w個Actor！其并發性能超強悍~~~

Akka官方文檔:http://akka.io/docs/

背景知識簡要介紹完了，本篇的主題是怎樣實作Akka文檔中介紹的router

router即路由，消息傳遞到router後，router根據響應的政策将消息下發到其所管理的routees, routees可以看做是一系列Actor的集合，每個Actor當然既可以是本地的Actor也可以是遠端的Actor。本文将給出本地和遠端的兩種測試執行個體。

準備工作:

開發環境:ItelliJ IDEA

工程類型:Maven

1.工程目錄結構

初識Akka之Router

2.Maven 配置檔案

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.akka.route</groupId>
    <artifactId>akka</artifactId>
    <version>1.0-SNAPSHOT</version>
    <name>${project.artifactId}-${project.version}</name>

    <repositories>
        <repository>
            <id>akka-snapshots</id>
            <name>Akka Snapshots</name>
            <url>http://repo.akka.io/snapshots/</url>
            <layout>default</layout>
        </repository>
    </repositories>

    <dependencies>
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-actor_2.11</artifactId>
            <!--Must be 2.2-SNAPSHOT-->
            <version>2.3.12</version>
        </dependency>
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-remote_2.11</artifactId>
            <version>2.3.12</version>
        </dependency>
        <dependency>
            <groupId>org.apache.maven.archetypes</groupId>
            <artifactId>maven-archetype-quickstart</artifactId>
            <version>1.1</version>
        </dependency>

    </dependencies>

</project>

此項目中用到的akka版本是2.3.12，對應的遠端子產品remote版本也是2.3.12

3.akka配置檔案 application.conf

# akka 2.3
LocalSys{
  akka {
    actor {
      provider = "akka.remote.RemoteActorRefProvider"

      deployment {
        /router1{
          router = round-robin-pool
          nr-of-instances = 5

        }

        /router3{
          router = random-group
          routees.paths = [
            #set the remote actor path
            "akka.tcp://[email protected]:2552/user/remoteActor",
            #set the localhost actor path
            "akka.tcp://[email protected]:2552/user/remoteActor"
          ]
        }

      }
    }
  }
}

Akka的配置檔案很複雜，此處隻給出了需要的配置。

3.Main 程式

import akka.actor.*;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import com.typesafe.config.ConfigFactory;

/**
 * Created by rootigo on 2016/7/28.
 */
public class Main {
   /**For a remote akka project ,the top system must be set once,otherwise it will cause "Address already in use" exception,
   *The akka use default port 2552
   */
    public static ActorSystem _system = ActorSystem.create("parent", ConfigFactory
            .load().getConfig("LocalSys"));
    public static void main(String[] args) throws InterruptedException {
       ActorRef masterRef =
               _system.actorOf(Props.create(Master.class),"Master");
       for(int i = 0 ;i < 10;i ++) {
           int payload = i * 10;
           masterRef.tell(new Work(""+payload+""), null);
           Thread.sleep(1000 * 2);
       }


   }

}

Main是此項目的啟動程式，先根據配置檔案application.conf建立ActorSystem,後面由此system建立的Actor都在此system根路徑之下，即/parent/user/some_actor

4.Work檔案

import java.io.Serializable;
import java.util.Arrays;
import java.util.List;

import akka.actor.*;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.routing.*;
import com.typesafe.config.ConfigFactory;


public final class Work implements Serializable{
    private static final long serialVersionUID = 1L;
    public String payload;
    public Work(String payload){
        this.payload = payload;
    }
}
     /** Route Logic
       * RoundRobinRoutingLogic
       * RandomRoutingLogic
       * SmallestMailboxRoutingLogic
       * BroadcastRoutingLogic
       * ScatterGatherFirstCompletedRoutingLogic
       * TailChoppingRoutingLogic
       * ConsistentHashRoutingLogic
       * */
class Master extends UntypedActor {
    LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    ActorRef getRouteeRef(){
      
        ActorRef workerpool = Main._system.actorOf(Props.create(Worker.class)
        );
        return workerpool;
    }
    ActorRef getRemoteRouter(String remoteAddr){
      
        String remoteActorAddr = String.format("akka.tcp://[email protected]%s/user/remoteActor",remoteAddr);
        ActorRef routerRemote = getContext().actorFor(remoteActorAddr);
        return routerRemote;

    }
    ActorRef getRoundRobinPoolRef(){
        //Props.create(Worker.class) means create a Worker actor
        ActorRef routerRef = Main._system.actorOf(FromConfig.getInstance().props(Props.create(Worker.class)),"router1");
        System.out.printf("RoundRobinGroup actor path is:{%s}\n",routerRef.path());
        return routerRef;
    }
    ActorRef getRoundRobinGroupRefByCoding(){
        //Before using RoundRobinGroup(paths) to router w1,w2,w3,this three actors must be created
        Main._system.actorOf(Props.create(Worker.class),"w1");
        Main._system.actorOf(Props.create(Worker.class),"w2");
        Main._system.actorOf(Props.create(Worker.class),"w3");

        ///Round-robin-Group by program code
        List<String> paths = Arrays.asList("/user/w1", "/user/w2",
                "/user/w3");
        ActorRef routerRef = Main._system.actorOf(new RoundRobinGroup(paths).props(), "router2");
        ActorRef w1 = Main._system.actorFor("/user/w1");
        ActorRef w2 = Main._system.actorFor("/user/w2");
        ActorRef w3 = Main._system.actorFor("/user/w3");
        System.out.printf("W1 path:{%s,%s},W2 path:{%s,%s},W3 path:{%s,%s}\n",
                w1.path(),w1.isTerminated(),w2.path(),w2.isTerminated(),w3.path(),w3.isTerminated());
         /// Round-robin-group
         System.out.printf("RoundRobinGroup actor path is:{%s}\n",routerRef.path());
         return routerRef;

    }
    ActorRef getRoundRobinGroupRefByConfigureFile(){
        //router the remote actor as routees,make sure that the remote actors must be running
        ActorRef routerRef = Main._system.actorOf(FromConfig.getInstance().props(),"router3");
        System.out.printf("RoundRobinGroup actor path is:{%s}\n",routerRef.path());
        return routerRef;
    }
    void addRouteeProgramtically(List<Routee> routees){
        for(int i = 0 ;i < 5;i++){
            // ActorRef r = getContext().actorOf(Props.create(Worker.class),"Worker" + i);
            String remoteAddr = "192.168.86.129:2552";
            if(i % 2 ==0){
                remoteAddr = "127.0.0.1:2552";
            }
            ActorRef r = getRemoteRouter(remoteAddr);
            System.out.printf("{%s} path is {%s}\n",i,r.path());
            getContext().watch(r);
            routees.add(new ActorRefRoutee(r));
        }

    }
    //1.Use router and routees to router the message
    Router router ;
    /*
    {
        List<Routee> routees = new ArrayList<Routee>();
        ActorRef r = getRoundRobinGroupRefByConfigureFile();
        getContext().watch(r);
        routees.add(new ActorRefRoutee(r));
        router = new Router(new RoundRobinRoutingLogic(),routees);
    }
    */
    //2.Use the RoundRobinGroup or RoundRobinPool to get the router
    ActorPath routerPath =  getRoundRobinGroupRefByConfigureFile().path();
    public void onReceive(Object msg){
        ActorRef sender = getSender();
        if(msg instanceof String){
            log.info("Msg From Sender[{}]:{}\n",sender.path(),msg);
        }
        else if(msg instanceof Work){
            log.info("[{}]Send Forward Message:'{}' to router \n" ,getSelf().path(), ((Work) msg).payload,routerPath);
            /*
            //Sending messages via the router,the sender is the router iteself
            router.route(((Work) msg).payload,getSelf());
            */
            ActorSelection selection = getContext().actorSelection(routerPath);
            selection.tell(((Work) msg).payload,getSelf());

        } else if(msg instanceof Terminated){
            router = router.removeRoutee(((Terminated)msg).actor());
            ActorRef r = getContext().actorOf(Props.create(Worker.class));
            getContext().watch(r);
            router = router.addRoutee(new ActorRefRoutee(r));
        }
    }


}

需要強調的是，在用RoundRobinGroup(和其他Group)擷取作為router的Actor時，一定要保證routees.path中涉及的遠端Actor已經存在，否則消息無法正确傳遞。

方法getRoundRobinGroupByCoding()中的/user/w1,/user/w2,/user/w3三個Actor要提前建立。

從代碼中可以看出，建立Router有兩種方法，一種是通過向Router中逐個添加routee，另外一種就是用RoundRobinGroup或者RoudRobinPool來建立Router的ActorRef。

5.Worker檔案

import akka.actor.ActorPath;
import akka.actor.UntypedActor;
import akka.event.Logging;
import akka.event.LoggingAdapter;

public class Worker extends UntypedActor {
    LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    @Override
    public void onReceive(Object msg){
        ActorPath path = getSender().path();
        ActorPath myself = getSelf().path();
        if(msg instanceof  Work){
            String payload = ((Work)msg).payload;
            log.info("Receive Work({}) From :{}",payload,path);
        } else if(msg instanceof  String){
            log.info("Receive message:{} From :{}",msg,path);
        }

    }
}

最後，還需要一個遠端的服務端RemoteNodeApp，不再列出，下面給出附件，自行下載下傳。

http://download.csdn.net/detail/igoqhb/9590561

初識Akka之Router

繼續閱讀

grails2.3.X無法debug

微軟蘋果21大創新技術太前衛遭遇失敗

微軟和蘋果的21大創新技術太前衛遭遇失敗

.NET 事務處理總結

pycharm remote deployment 配置問題

WINDOWS DNA: 為網際網路時代建立 WINDOWS 應用WINDOWS DNA: 為網際網路時代建立 WINDOWS 應用

webservice 入門講解

動态Web services

soap和web services

關于 web service 詳解

為 HACMP/XD-PPRC 自動生成配置資料

ICE和ACE高效網絡程式設計中間件平台

COM技術内幕-讀書筆記-系列之一

對等點如何彼此定位

Erlang不能錯過的盛宴

中間件ICE介紹