AKKA Router路由

路由概念

大量的actor在并行工作的时候，处理到来的消息流，这时候就需要一个组件或者东西来引导消息从源到目的地Actor,这个组件或者东西就是Router
在Akka中，router也是一种actor 类型，它路由到来的消息到其他的actors,其他那些actors就叫做routees(被路由对象)

Routing模式由Router和Routee组成：
Routee是负责具体运算的Actor
Router 是把外界发来消息按照某种指定的方式（默认提供了多种路由逻辑类）分配给Routee去运算，用于调度actor任务或进行负载均衡策略

Pool 和 Group模式

Router又可分Pool和Group两种模式：
Router-Pool模式中Router负责构建所有的Routee。路由器创建routee作为子actor，并在该子actor终止时将它从路由器中移除。

Router-Group模式中的Routees由外界其它Actor产生(自行创建,自行管理)，特点是能实现灵活的Routee构建和监控

路由逻辑类

akka.routing.RoundRobinRoutingLogic  轮询
akka.routing.RandomRoutingLogic  随机
akka.routing.SmallestMailboxRoutingLogic 空闲
akka.routing.BroadcastRoutingLogic 广播
akka.routing.ScatterGatherFirstCompletedRoutingLogic
akka.routing.TailChoppingRoutingLogic
akka.routing.ConsistentHashingRoutingLogic

实现示例：

package routerimport akka.actor.AbstractActor
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Props
import akka.japi.pf.ReceiveBuilder
import akka.routing.*
import org.slf4j.LoggerFactory/*** Created by: tankx* Date: 2019/7/20* Description: 路由示例*/
class WorkActor : AbstractActor() {var log = LoggerFactory.getLogger(WorkActor::class.java)override fun createReceive(): Receive {return ReceiveBuilder.create().matchAny(this::receive).build()}fun receive(msg: Any) {log.info(" {}: $msg", self.path())}
}class RouterActor : AbstractActor() {var log = LoggerFactory.getLogger(RouterActor::class.java)private lateinit var router: Router;override fun preStart() {super.preStart()var list = arrayListOf<Routee>()for (i in 1..10) {var worker = context.actorOf(Props.create(WorkActor::class.java), "worker_$i")list.add(ActorRefRoutee(worker))}/*** 路由方式* RoundRobinRoutingLogic: 轮询* BroadcastRoutingLogic: 广播* RandomRoutingLogic: 随机* SmallestMailboxRoutingLogic: 空闲*/router = Router(RoundRobinRoutingLogic(), list)}override fun createReceive(): Receive {return ReceiveBuilder.create().matchAny(this::receive).build()}fun receive(msg: Any) {//log.info("RouterActor : $msg")
    router.route(msg, sender)}}fun main() {var actorSystem = ActorSystem.create("RouterSystem")var routerActor = actorSystem.actorOf(Props.create(RouterActor::class.java))for (i in 1..20) {Thread.sleep(2000)routerActor.tell("消息来了", ActorRef.noSender())}}

指定path模式，远程调用（集群内部访问）

package router.clusterimport akka.actor.AbstractActor
import akka.actor.Props
import akka.japi.pf.ReceiveBuilder
import org.slf4j.LoggerFactory
import router.WorkActor/*** Created by: tankx* Date: 2019/7/20* Description:*/
class WorkActor : AbstractActor() {var log = LoggerFactory.getLogger(WorkActor::class.java)override fun createReceive(): Receive {return ReceiveBuilder.create().matchAny(this::receive).build()}fun receive(msg: Any) {log.info(" {}: $msg", self.path())}}

package router.clusterimport akka.actor.*
import akka.japi.pf.ReceiveBuilder
import org.slf4j.LoggerFactory
import router.WorkActor
import akka.remote.routing.RemoteRouterConfig
import akka.routing.BroadcastGroup
import akka.routing.BroadcastPool
import akka.routing.FromConfig
import akka.routing.RoundRobinPool
import com.typesafe.config.ConfigFactory/*** Created by: tankx* Date: 2019/7/20* Description:*/
class MyService : AbstractActor() {var log = LoggerFactory.getLogger(MyService::class.java)override fun createReceive(): Receive {return ReceiveBuilder.create().matchAny(this::receive).build()}fun receive(msg: Any) {log.info(" {}: $msg", self.path())}}fun getActorSystem(port: Int): ActorSystem {val config = ConfigFactory.parseString("akka.remote.netty.tcp.port=$port").withFallback(ConfigFactory.load("application_router.conf"))var actorSystem = ActorSystem.create("RouterSystem", config);return actorSystem
}
//读取配置文件方式
fun loadRouterByConfig() {val actorSystem = getActorSystem(3662);var router = actorSystem.actorOf(FromConfig.getInstance().props(), "workerRouter")for (i in 1..10) {router.tell("test", ActorRef.noSender())Thread.sleep(2000)}
}
//代码方式
fun loadRouterByCode() {val actorSystem = getActorSystem(3662);var address = listOf<Address>(AddressFromURIString.parse("akka.tcp://RouterSystem@127.0.0.1:2663"),AddressFromURIString.parse("akka.tcp://RouterSystem@127.0.0.1:2661"))//var paths = listOf("/user/MyWorker")//可以配置多个地址 akka://RouterSystem/user/MyWorker
var router = actorSystem.actorOf(RemoteRouterConfig(BroadcastPool(2),address).props(Props.create(router.cluster.WorkActor::class.java)), "workerRouter")for (i in 1..20) {router.tell("test", ActorRef.noSender())Thread.sleep(2000)}
}fun main() {//loadRouterByCode()
loadRouterByConfig()}

配置文件

akka {actor {provider = "cluster"}remote {//log-remote-lifecycle-events = off
    netty.tcp {hostname = "127.0.0.1"port = 2661}}# For the sample, just bind to loopback and do not allow access from the network# the port is overridden by the logic in main class//  remote.artery {//    enabled = on//    transport = tcp//    canonical.port = 2661//    canonical.hostname = 127.0.0.1//  }
cluster {seed-nodes = ["akka.tcp://RouterSystem@127.0.0.1:2661"]# auto downing is NOT safe for production deployments.# you may want to use it during development, read more about it in the docs.auto-down-unreachable-after = 10s}
}

分不同JVM启动

fun main() {var actorSystem = getActorSystem(2661)var workActor = actorSystem.actorOf(Props.create(WorkActor::class.java), "MyWorker")}

fun main() {var actorSystem = getActorSystem(2663)var workActor = actorSystem.actorOf(Props.create(WorkActor::class.java), "MyWorker")}

结果调用输出

2019-07-22 12:25:42.589 [RouterSystem-akka.actor.default-dispatcher-21] INFO  router.WorkActor -  akka://RouterSystem/remote/akka.tcp/RouterSystem@127.0.0.1:3662/user/workerRouter/c2: test
2019-07-22 12:25:44.589 [RouterSystem-akka.actor.default-dispatcher-2] INFO  router.WorkActor -  akka://RouterSystem/remote/akka.tcp/RouterSystem@127.0.0.1:3662/user/workerRouter/c2: test
2019-07-22 12:25:46.590 [RouterSystem-akka.actor.default-dispatcher-2] INFO  router.WorkActor -  akka://RouterSystem/remote/akka.tcp/RouterSystem@127.0.0.1:3662/user/workerRouter/c2: test

通过以上示例基本可以了解AKK的路由模式与应用了。

转载于:https://www.cnblogs.com/tankaixiong/p/11225259.html