初識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.工程目錄結構
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,不再列出,下面給出附件,自行下載。
相關推薦
初識Akka之Router
背景介紹 簡單來說Akka是一個用於構建服務端分散式應用的框架,其具有高併發、可擴充套件、容錯性等特點。Akka中的基本單位是Actor,Actor可以與執行緒來作為類比,有自己的識別符號、儲存空間、排程策略、生命週期等。每個Actor相互之間的互動都是通
初識Scrapy之再續火影情緣
復雜 功能 splay 註意 回調函數 正則 理解 turn bject 前言Scrapy框架之初窺門徑1 Scrapy簡介2 Scrapy安裝3 Scrapy基礎31 創建項目32 Shell分析4 Scrapy程序編寫41 Spiders程序測試42 Items編寫43
轉:vue-router 2.0 常用基礎知識點之router.push()
als 返回 ack name rep targe 代碼 添加 clas 轉載地址:http://www.jianshu.com/p/ee7ff3d1d93d router.push(location) 除了使用 <router-link> 創建 a 標簽來
初識Hibernate之關聯映射(一)
ber 初識 album nat amp uid 關聯映射 映射 pic http://pic.cnhubei.com/space.php?uid=1774&do=album&id=1361989http://pic.cnhubei.com/space.ph
初識linux之文件管理
當前 head 修改文件內容 文本編輯 director set 創建目錄 選項 一行 文件管理一、文件基本信息:時間1.stat 後面加文件可以查看該文件的最近訪問,更改,改動的時間以及環境,權限等 atime 是訪問時間mtime是修改時間,改內容ctime是更改時間,
初識linux之暢遊其中
-- 一個 themes 詳細 修改 file type at命令 類型 1.文件基本信息 時間類型:訪問時間、更改時間、改動時間、刪除時間 可以使用stat命令查看文件時間 2.文件類型 -普通文件 d目錄文件 b設備文件(塊設備)
初識linux之自己的家
chage -c inux 追加 附加 查看 管理 nbsp 附加組 用戶管理 Linux系統是一個多用戶多任務的分時操作系統,任何一個要使用系統資源的用戶,都必須首先向系統管理員申請一個賬號,然後以這個賬號的身份進入系統。 用戶的賬號一方面可以幫助系統管理員對使用
初識linux之給我一個家
don 名稱 收回 內容 個人 永久 su - 初識linux ash 用戶管理1.用戶/組基本概念查看當前登錄的用戶信息:id查看文件的owner:LL可以看到文件的所屬者查看運行進程:ps aux | less,分屏顯示當前進程用戶和組的存儲文件 /etc/pass
初識TomCat之2——TomCat介紹及安裝
重啟 一個tomcat security direct 查看 介紹 虛擬 vpd use 一、TomCat介紹通過之前的介紹,Tomcat與JDK共同構成了一個Web Container容器,它在JDK的基礎上提供了Servlet和JSP組件,可以接受來自客戶端的動態請求,
初識TomCat之3——TomCat動靜分離
tomcat 動靜分離 一、背景介紹在生產環境中tomcat很少直接用於響應客戶端請求,即使它可以通過Connector組件接收用戶的請求,一般來說常見於server.xml配置文件中Connector連接器的類型有3種:http、SSL和AJP 1.3在生產環境中,tomcat前端一般會有一個專門的
初識TomCat之4——TomCat負載均衡及基於IP的Session sticky
tomcat Session sticky 一、背景介紹當一個tomcat服務器性能遇到瓶頸時,通常的做法是將tomcat服務器橫向擴容,使用多臺tomcat服務器來響應用戶請求,但此時又引入了一個新問題,就是用戶每次請求都會隨機給他分配一個tomcat服務器,下面介紹如何實現tomcat實現負載均衡
初識MariaDB之4——Xtrabackup備份與還原
mysql mariadb xtrabackup 一、基礎知識mysqldump是mysql自帶的一個單線程的邏輯備份工具,當數據量很大時,使用mysqldump備份的時間會非常長,且無法做增量備份,xtrabackup是一款基於底層塊復制、開源的備份工具,可以對數據庫在線實現備份。xtraba
初識MariaDB之5——主從復制原理及部署
mysql mariadb 主從復制 一、原理簡介當前主流的數據庫為了保證數據一致性,都采用支持事務類型的數據存儲引擎,如innodb等,當一個事務提交後先寫到事務日誌中,在進行數據修改,並且將對數據產生任何影響的語句都記錄在二進制日誌中,主從復制就是基於二進制日誌實現的。主從同步的工作流程為:1
初識MariaDB之1——MariaDB的安裝
mysql mariadb 一、MariaDB介紹MariaDB是Mysql在被Oracle收購後由原作者重新開發的一種關系型數據庫程序,由於是同一個作者,所以在操作上與Mysql十分類似,版本也是相互對應,如:MariaDB5.5對應Mysql的5.5版本,但從Mysql5.6開始對應的MariaD
初識MariaDB之6——半同步復制
mysql mariadb 半同步 一、背景介紹MySQL主從復制能解決一定的單點故障問題,但其異步的工作特性存在一定的隱患,比如主節點事務提交後還未寫入binlog,此時主節點故障後,但主節點認為事務已提交,從節點無法從主節點獲取到該事務的信息,提升為主節點後就會造成數據丟失,為了解決這一問題,
初識MariaDB之8——GTID主從復制
mysql mariadb gtid 一、背景介紹在MySQL5.6之前,主從復制是通過binlog和position實現的,當A主機宕機後,B主機成為新的主節點,此時在C主機上需要使用sql語句:CHANGE MASTER TO MASTER_HOST='xxx', MASTER
初識MariaDB之9——keepalive+GTID雙主實現高可用
mysql mariadb gtid 一、背景介紹傳統的主從復制架構下,盡管可以使用半同步的方式進行讀寫分離,但是主節點仍然存在單點隱患,在規模不大的情況下可以采用keepalive+雙主的模式對主節點進行高可用保護,客戶端通過VIP訪問MySQL服務器二、實現方式keepalive是通過vrrp
初識MariaDB之7——MHA高可用
mysql mariadb mha 一、背景介紹在傳統的MySQL一主一從或一主多從結構中,主節點存在單點故障隱患,為了解決該故障,進而引入了MHA高可用的方式。MHA由manager和node兩部分組成,manager可以安裝在一個獨立的服務器上或非master角色的MySQL服務器上,1個ma
初識MariaDB之10——MariaDB Galera Cluster
mysql mariadb PXC Galera 一、背景介紹無論是采用binlog或者GTID的方式,其本質都是通過I/O_thread和sql_thread的形式進行的同步,因為無法避免復制延遲而飽受詬病,基於上述MariaDB引入了Galera Cluster來解決此問題。二、Galer
初識C#之變量
關鍵字 數據類型 不能 第一個 數字 初識 變量存儲 定義 合並 變量:一塊存儲的內存空間,且該內存區域的數據內容可以變化。在代碼中使用變量,變量必須先聲明後賦值。 語法:數據類型 變量名; 賦值:變量名=數據; 變量用於在內存中存儲數據、變量存儲的數據可變