Akka之消息路由

妖狐艹你老母 2022-05-12 05:28 244阅读 0赞

> Akka提供了非常灵活的消息发送机制。有时候我们使用一组Actor来提供服务，这一组Actor中所有的Actor都是对等的，也就是说你可以找任何一个Actor来为你服务。这种情况下，如何才能快速有效地找到合适的Actor呢？或者说如何调度这些消息，才可以使负载更为均衡地分配在这一组Actor中呢？为了解决这个问题，Akka使用一个路由器组件（Router）来封装消息的调度。系统提供了几种实用的消息路由策略，比如，轮询选择Actor进行消息发送，随机消息发送，将消息发送给最为空闲的Actor，甚至是在组内广播消息。

## 实例 ##

package com.bzb.java8.akka;
    
    import akka.actor.UntypedActor;
    import akka.event.Logging;
    import akka.event.LoggingAdapter;
    
    /** * @author bzb * @Description: 当前示例中的唯一一个Actor * @date 2018/9/14 14:16 */
    public class MyWorker extends UntypedActor { 
    
        private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    
        public static enum Msg {
            WORKING, DONE, CLOSE;
        }
    
        @Override
        public void onReceive(Object msg) throws Exception {
            if (msg == Msg.WORKING) {
                System.out.println("I am working");
            }
            if (msg == Msg.DONE) {
                System.out.println("Stop working");
            }
            if (msg == Msg.CLOSE) {
                System.out.println("I will shutdow");
    
                getSender().tell(Msg.CLOSE, getSelf());
    
                // 收到CLOSE消息时，关闭自己，结束运行。
                getContext().stop(getSelf());
            } else {
                unhandled(msg);
            }
        }
    }

package com.bzb.java8.akka;
    
    import akka.actor.ActorRef;
    import akka.actor.Props;
    import akka.actor.Terminated;
    import akka.actor.UntypedActor;
    import akka.event.Logging;
    import akka.event.LoggingAdapter;
    import akka.routing.ActorRefRoutee;
    import akka.routing.RoundRobinRoutingLogic;
    import akka.routing.Routee;
    import akka.routing.Router;
    
    import java.util.ArrayList;
    import java.util.List;
    
    /** * @author bzb * @Description: 监视者，如同劳动监工一样，一旦被监视者因为停止工作，则监视者就会收到一条消息 * @date 2018/9/14 14:27 */
    public class WatchActor extends UntypedActor { 
    
        private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
    
        // 定义路由器组件Router
        public Router router;
    
        {
            List<Routee> routees = new ArrayList<>();
            for (int i = 0; i < 5; i++) {
                ActorRef worker = getContext().actorOf(Props.create(MyWorker.class), "worker-" + i);
                getContext().watch(worker);
                routees.add(new ActorRefRoutee(worker));
            }
    
            // 构造Router时，需要指定路由策略和一组被路由的Actor(Routee)。这里使用了RoundRobinRoutingLogin，也就是对所有的Routee轮询发送消息。
            /** * RoundRobinRoutingLogic: 轮询 * BroadcastRoutingLogic: 广播 * RandomRoutingLogic: 随机 * SmallestMailboxRoutingLogic: 空闲 */
            router = new Router(new RoundRobinRoutingLogic(), routees);
        }
    
        @Override
        public void onReceive(Object msg) throws Exception {
            if (msg instanceof MyWorker.Msg) {
                // 当需要投递消息给这5个worker时，只需要将消息投递给这个Router即可。
                // Router给根据给定的路由策略进行消息投递。
                router.route(msg, getSender());
            } else if (msg instanceof Terminated) { // 终止消息
    
                // 当一个worker停止工作时，可以简单的将其从工作组中移除
                router = router.removeRoutee(((Terminated) msg).actor());
    
                // 打印终止的Actor路径和剩余工作组的大小
                System.out.println(((Terminated) msg).actor().path() + " is closed, routees = " + router.routees().size());
    
                // 如果系统中没有可以的Actor，就会直接关闭系统
                if (router.routees().size() == 0) {
                    System.out.println("Closed Systme");
                    RouteMain.flag.send(false);
                    getContext().system().shutdown();
                }
             } else {
                unhandled(msg);
            }
        }
    }

package com.bzb.java8.akka;
    
    
    import akka.actor.ActorRef;
    import akka.actor.ActorSystem;
    import akka.actor.Props;
    import akka.agent.Agent;
    import akka.dispatch.ExecutionContexts;
    import com.typesafe.config.ConfigFactory;
    import scala.concurrent.ExecutionContext;
    
    /** * @author bzb * @Description: * @date 2018/9/14 17:42 */
    public class RouteMain { 
    
        public static Agent<Boolean> flag = Agent.create(true, ExecutionContexts.global());
    
        public static void main(String[] args) throws InterruptedException {
            ActorSystem system = ActorSystem.create("route", ConfigFactory.load("samplehello.conf"));
    
            ActorRef watcher = system.actorOf(Props.create(WatchActor.class), "watcher");
    
            int i = 1;
    
            // 不停发送消息，知道flag为false，即路由绑定的工作组中的Actor为0
            while (flag.get()) {
                watcher.tell(MyWorker.Msg.WORKING, ActorRef.noSender());
                if (i % 10 == 0) {
                    watcher.tell(MyWorker.Msg.CLOSE, ActorRef.noSender()); // 每10次发送一条关闭消息
                }
                i++;
                Thread.sleep(100);
            }
        }
    }

## **参考资源** ##

**<实战java高并发程序设计>**