akka学习教程(十四) akka分布式实战

灰太狼 2022-07-11 06:57 435阅读 0赞

# akka系列文章目录 #

*  [akka学习教程(十四) akka分布式实战][akka_ akka]
 *  [akka学习教程(十三) akka分布式][akka_ akka 1]
 *  [akka学习教程(十二) Spring与Akka的集成][akka_ Spring_Akka]
 *  [akka学习教程(十一) akka持久化][akka_ akka 2]
 *  [akka学习教程(十) agent][akka_ agent]
 *  [akka学习教程(九) STM软件事务内存][akka_ STM]
 *  [akka学习教程(八) Actor中的Future-询问模式][akka_ Actor_Future-]
 *  [akka学习教程(七) 内置状态转换Procedure][akka_ _Procedure]
 *  [akka学习教程(六) 路由器Router][akka_ _Router]
 *  [akka学习教程(五) inbox消息收件箱][akka_ inbox]
 *  [akka学习教程(四) actor生命周期][akka_ actor]
 *  [akka学习教程(三) 不可变对象][akka_]
 *  [akka学习教程(二)HelloWord][akka_HelloWord]
 *  [akka学习教程(一)简介][akka]

> 上一篇文章介绍了akka集群的搭建，现在假如服务的生产者与消费者两个角色，模拟真实的服务调用。  
> 本篇文章主要参考 [使用Akka构建集群（二）][Akka]

# 整体架构 #

服务端三个服务，端口为2552,2553,2551；客户端有两个：2554,2555  
服务端角色为\[server\]；客户端角色为\[client\]

# 服务端 #

## 集群角色 ##

首先配置服务端集群角色为\[server\]:

akka {
      loglevel = "INFO"
    
      actor {
          provider = "akka.cluster.ClusterActorRefProvider"
        }
        remote {
          log-remote-lifecycle-events = off
          netty.tcp {
            hostname = "127.0.0.1"
            port = 2551
          }
        }
    
        cluster {
          seed-nodes = [
            "akka.tcp://akkaClusterTest@127.0.0.1:2551",
            "akka.tcp://akkaClusterTest@127.0.0.1:2552"]
    
          #//#snippet
          # excluded from snippet
          auto-down-unreachable-after = 10s
          #//#snippet
          # 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
    
          roles = [server]
          # Disable legacy metrics in akka-cluster.
          metrics.enabled=off
        }
    }
    
    # 持久化相关
    akka.persistence.journal.plugin = "akka.persistence.journal.inmem"
    # Absolute path to the default snapshot store plugin configuration entry.
    akka.persistence.snapshot-store.plugin = "akka.persistence.snapshot-store.local"

## 定义通讯的数据结构 ##

> 实际开发过程中，可以将该数据结构作为单独的接口供服务端和客户端引用。

package akka.myCluster;
    
    import java.io.Serializable;
    
    public class TransformationMessages { 
    
        /** * 传递的数据（参数） */
        public static class TransformationJob implements Serializable { 
            private final String text;  
    
            public TransformationJob(String text) {  
                this.text = text;  
            }  
    
            public String getText() {  
                return text;  
            }  
        }
    
        /** * 返回结果 */
        public static class TransformationResult implements Serializable {   
            private final String text;  
    
            public TransformationResult(String text) {  
                this.text = text;  
            }  
    
            public String getText() {  
                return text;  
            }  
    
            @Override  
            public String toString() {  
                return "TransformationResult(" + text + ")";  
            }  
        }
    
        /** * 异常处理 */
        public static class JobFailed implements Serializable {   
            private final String reason;  
            private final TransformationJob job;  
    
            public JobFailed(String reason, TransformationJob job) {  
                this.reason = reason;  
                this.job = job;  
            }  
    
            public String getReason() {  
                return reason;  
            }  
    
            public TransformationJob getJob() {  
                return job;  
            }  
    
            @Override  
            public String toString() {  
                return "JobFailed(" + reason + ")";  
            }  
        }  
    
        /** * 用于服务端向客户端注册 */
        public static final int BACKEND_REGISTRATION = 1;
    
    }

真正服务端业务逻辑处理代码：**简单讲收到的字符串转为大写返回**

package akka.myCluster;
    
    import akka.actor.ActorSystem;
    import akka.actor.Props;
    import akka.actor.UntypedActor;
    import akka.cluster.Cluster;
    import akka.cluster.ClusterEvent;
    import akka.cluster.Member;
    import akka.cluster.MemberStatus;
    import akka.event.Logging;
    import akka.event.LoggingAdapter;
    import com.typesafe.config.ConfigFactory;
    
    import static akka.myCluster.TransformationMessages.BACKEND_REGISTRATION;
    
    public class MyAkkaClusterServer extends UntypedActor { 
    
        LoggingAdapter logger = Logging.getLogger(getContext().system(), this);
    
        Cluster cluster = Cluster.get(getContext().system());
    
        // subscribe to cluster changes 
        @Override  
        public void preStart() {  
            // #subscribe 
            cluster.subscribe(getSelf(), ClusterEvent.MemberUp.class);
            // #subscribe 
        }  
    
        // re-subscribe when restart 
        @Override  
        public void postStop() {  
            cluster.unsubscribe(getSelf());  
        }  
    
        @Override  
        public void onReceive(Object message) {
            if (message instanceof TransformationMessages.TransformationJob) {
                TransformationMessages.TransformationJob job = (TransformationMessages.TransformationJob) message;
                logger.info(job.getText());
                getSender().tell(new TransformationMessages.TransformationResult(job.getText().toUpperCase()), getSelf());
    
            } else if (message instanceof ClusterEvent.CurrentClusterState) {
                /** * 当前节点在刚刚加入集群时，会收到CurrentClusterState消息，从中可以解析出集群中的所有前端节点（即roles为frontend的），并向其发送BACKEND_REGISTRATION消息，用于注册自己 */
                ClusterEvent.CurrentClusterState state = (ClusterEvent.CurrentClusterState) message;
                for (Member member : state.getMembers()) {
                    if (member.status().equals(MemberStatus.up())) {
                        register(member);
                    }
                }
    
            } else if (message instanceof ClusterEvent.MemberUp) {
                /** * 有新的节点加入 */
                ClusterEvent.MemberUp mUp = (ClusterEvent.MemberUp) message;
                register(mUp.member());
    
            } else {
                unhandled(message);
            }
    
        }
    
        /** * 如果是客户端角色，则像客户端注册自己的信息。客户端收到消息以后会讲这个服务端存到本机服务列表中 * @param member */
        void register(Member member) {
            if (member.hasRole("client"))
                getContext().actorSelection(member.address() + "/user/myAkkaClusterClient").tell(BACKEND_REGISTRATION, getSelf());
        }
    
        public static void main(String [] args){
            System.out.println("Start MyAkkaClusterServer");
            ActorSystem system = ActorSystem.create("akkaClusterTest", ConfigFactory.load("reference.conf"));
            system.actorOf(Props.create(MyAkkaClusterServer.class), "myAkkaClusterServer");
            System.out.println("Started MyAkkaClusterServer");
    
        }
    }

## 如何保证服务发现与维护 ##

上面代码已经有注释了，有2点：  
\- 有新节点加入时，如果是客户端角色，则像客户端注册自己的信息。客户端收到消息以后会讲这个服务端存到本机服务列表中  
\- 服务端当前节点在刚刚加入集群时，会收到CurrentClusterState消息，从中可以解析出集群中的所有前端节点（即roles为frontend的），并向其发送BACKEND\_REGISTRATION消息，用于注册自己

# 客户端 #

修改客户端roles为client

akka {
      loglevel = "INFO"
    
      actor {
          provider = "akka.cluster.ClusterActorRefProvider"
        }
        remote {
          log-remote-lifecycle-events = off
          netty.tcp {
            hostname = "127.0.0.1"
            port = 2554
          }
        }
    
        cluster {
          seed-nodes = [
            "akka.tcp://akkaClusterTest@127.0.0.1:2551",
            "akka.tcp://akkaClusterTest@127.0.0.1:2552"]
    
          #//#snippet
          # excluded from snippet
          auto-down-unreachable-after = 10s
          #//#snippet
          # 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
    
        roles = [client]
          # Disable legacy metrics in akka-cluster.
          metrics.enabled=off
        }
    }
    
    # 持久化相关
    akka.persistence.journal.plugin = "akka.persistence.journal.inmem"
    # Absolute path to the default snapshot store plugin configuration entry.
    akka.persistence.snapshot-store.plugin = "akka.persistence.snapshot-store.local"

客户端代码

package akka.myCluster;
    
    import akka.actor.*;
    import akka.dispatch.OnSuccess;
    import akka.util.Timeout;
    import com.typesafe.config.ConfigFactory;
    import scala.concurrent.ExecutionContext;
    import scala.concurrent.duration.Duration;
    import scala.concurrent.duration.FiniteDuration;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    import static akka.myCluster.TransformationMessages.BACKEND_REGISTRATION;
    import static akka.pattern.Patterns.ask;
    
    public class MyAkkaClusterClient extends UntypedActor { 
    
        List<ActorRef> backends = new ArrayList<ActorRef>();
        int jobCounter = 0;
    
        @Override
        public void onReceive(Object message) {
            if ((message instanceof TransformationMessages.TransformationJob) && backends.isEmpty()) {
       //无服务提供者
                TransformationMessages.TransformationJob job = (TransformationMessages.TransformationJob) message;
                getSender().tell(
                        new TransformationMessages.JobFailed("Service unavailable, try again later", job),
                        getSender());
    
            } else if (message instanceof TransformationMessages.TransformationJob) {
                TransformationMessages.TransformationJob job = (TransformationMessages.TransformationJob) message;
                /** * 这里在客户端业务代码里进行负载均衡操作。实际业务中可以提供多种负载均衡策略，并且也可以做分流限流等各种控制。 */
                jobCounter++;
                backends.get(jobCounter % backends.size())
                        .forward(job, getContext());
    
            } else if (message == BACKEND_REGISTRATION) {
                /** * 注册服务提供者 */
                getContext().watch(getSender());//这里对服务提供者进行watch
                backends.add(getSender());
    
            } else if (message instanceof Terminated) {
                /** * 移除服务提供者 */
                Terminated terminated = (Terminated) message;
                backends.remove(terminated.getActor());
    
            } else {
                unhandled(message);
            }
        }
    
        public static void main(String [] args){
            System.out.println("Start myAkkaClusterClient");
            ActorSystem actorSystem = ActorSystem.create("akkaClusterTest", ConfigFactory.load("reference.conf"));
            final ActorRef myAkkaClusterClient = actorSystem.actorOf(Props.create(MyAkkaClusterClient.class), "myAkkaClusterClient");
            System.out.println("Started myAkkaClusterClient");
    
            final FiniteDuration interval = Duration.create(2, TimeUnit.SECONDS);
            final Timeout timeout = new Timeout(Duration.create(5, TimeUnit.SECONDS));
            final ExecutionContext ec = actorSystem.dispatcher();
            final AtomicInteger counter = new AtomicInteger();
    
            actorSystem.scheduler().schedule(interval, interval, new Runnable() {
                public void run() {
                    ask(myAkkaClusterClient, new TransformationMessages.TransformationJob("hello-" + counter.incrementAndGet()), timeout)
                            .onSuccess(new OnSuccess<Object>() {
                                public void onSuccess(Object result) {
                                    System.out.println(result.toString());
                                }
                            }, ec);
                }
            }, ec);
    
        }
    }

可以看到TransformationFrontend处理的消息分为以下三种：

*  **BACKEND\_REGISTRATION**：收到此消息说明有服务端通知客户端，TransformationFrontend首先将服务端的ActorRef加入backends列表，然后对服务端的ActorRef添加监管；
 *  **Terminated**：由于TransformationFrontend对服务端的ActorRef添加了监管，所以当服务端进程奔溃或者重启时，将收到Terminated消息，此时TransformationFrontend将此服务端的ActorRef从backends列表中移除；
 *  **TransformationJob**：此消息说明有新的转换任务需要TransformationFrontend处理，处理分两种情况：  
    
    
     *  `backends`列表为空，则向发送此任务的发送者返回JobFailed消息，并告知“目前没有服务端可用，请稍后再试”；
     *  `backends`列表不为空，则通过取模运算选出一个服务端，将TransformationJob转发给服务端进一步处理；

# 运行结果 #

> 启动3个服务端，2个客户端

服务端2551输出：

[INFO] [01/18/2017 17:01:57.167] [akkaClusterTest-akka.actor.default-dispatcher-20] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-4
    [INFO] [01/18/2017 17:02:02.438] [akkaClusterTest-akka.actor.default-dispatcher-20] [akka://akkaClusterTest/user/myAkkaClusterServer] hello.2555-4
    [INFO] [01/18/2017 17:02:03.124] [akkaClusterTest-akka.actor.default-dispatcher-3] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-7
    [INFO] [01/18/2017 17:02:08.416] [akkaClusterTest-akka.actor.default-dispatcher-20] [akka://akkaClusterTest/user/myAkkaClusterServer] hello.2555-7
    [INFO] [01/18/2017 17:02:09.137] [akkaClusterTest-akka.actor.default-dispatcher-18] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-10
    [INFO] [01/18/2017 17:02:14.414] [akkaClusterTest-akka.actor.default-dispatcher-17] [akka://akkaClusterTest/user/myAkkaClusterServer] hello.2555-10
    [WARN] [01/18/2017 17:02:15.104] [akkaClusterTest-akka.remote.default-remote-dispatcher-6] 
    [INFO] [01/18/2017 17:02:15.204] [akkaClusterTest-akka.actor.default-dispatcher-17] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-13

服务端2552输出：

[INFO] [01/18/2017 17:01:53.178] [akkaClusterTest-akka.actor.default-dispatcher-5] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-2
    [INFO] [01/18/2017 17:01:59.125] [akkaClusterTest-akka.actor.default-dispatcher-22] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-5
    [INFO] [01/18/2017 17:02:00.433] [akkaClusterTest-akka.actor.default-dispatcher-17] [akka://akkaClusterTest/user/myAkkaClusterServer] hello.2555-3
    [INFO] [01/18/2017 17:02:05.126] [akkaClusterTest-akka.actor.default-dispatcher-6] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-8
    [INFO] [01/18/2017 17:02:06.427] [akkaClusterTest-akka.actor.default-dispatcher-17] [akka://akkaClusterTest/user/myAkkaClusterServer] hello.2555-6
    [INFO] [01/18/2017 17:02:11.130] [akkaClusterTest-akka.actor.default-dispatcher-5] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-11
    [INFO] [01/18/2017 17:02:12.420] [akkaClusterTest-akka.actor.default-dispatcher-17] [akka://akkaClusterTest/user/myAkkaClusterServer] hello.2555-9
    [WARN] [01/18/2017 17:02:15.053] [akkaClusterTest-akka.remote.default-remote-dispatcher-7] 
    [INFO] [01/18/2017 17:02:17.123] [akkaClusterTest-akka.actor.default-dispatcher-16] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-14

服务端2553输出：

[INFO] [01/18/2017 17:01:55.144] [akkaClusterTest-akka.actor.default-dispatcher-20] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-3
    [INFO] [01/18/2017 17:01:58.428] [akkaClusterTest-akka.actor.default-dispatcher-4] [akka://akkaClusterTest/user/myAkkaClusterServer] hello.2555-2
    [INFO] [01/18/2017 17:02:01.130] [akkaClusterTest-akka.actor.default-dispatcher-16] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-6
    [INFO] [01/18/2017 17:02:04.413] [akkaClusterTest-akka.actor.default-dispatcher-16] [akka://akkaClusterTest/user/myAkkaClusterServer] hello.2555-5
    [INFO] [01/18/2017 17:02:07.141] [akkaClusterTest-akka.actor.default-dispatcher-5] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-9
    [INFO] [01/18/2017 17:02:10.413] [akkaClusterTest-akka.actor.default-dispatcher-18] [akka://akkaClusterTest/user/myAkkaClusterServer] hello.2555-8
    [INFO] [01/18/2017 17:02:13.128] [akkaClusterTest-akka.actor.default-dispatcher-18] [akka://akkaClusterTest/user/myAkkaClusterServer] hello-12

客户端2554输出

JobFailed(Service unavailable, try again later)
    TransformationResult(HELLO-2)
    TransformationResult(HELLO-3)
    TransformationResult(HELLO-4)
    TransformationResult(HELLO-5)
    TransformationResult(HELLO-6)
    TransformationResult(HELLO-7)
    TransformationResult(HELLO-8)
    TransformationResult(HELLO-9)
    TransformationResult(HELLO-10)
    TransformationResult(HELLO-11)
    TransformationResult(HELLO-12)

客户端2555输出

JobFailed(Service unavailable, try again later)
    TransformationResult(HELLO.2555-2)
    TransformationResult(HELLO.2555-3)
    TransformationResult(HELLO.2555-4)
    TransformationResult(HELLO.2555-5)
    TransformationResult(HELLO.2555-6)
    TransformationResult(HELLO.2555-7)
    TransformationResult(HELLO.2555-8)
    TransformationResult(HELLO.2555-9)
    TransformationResult(HELLO.2555-10)

可以发现，客户端发送的消息被服务端正确消费，并且进行了负载均衡。不过上面第一条消息由于客户端节点刚开始处理消息时，backends列表里还没有缓存好任何backend的ActorRef，所以报错`JobFailed(Service unavailable, try again later)`

# 总结 #

与thrift一样，使用akka需要自己进行服务的发现治理工作。但是著名的spark都完全依赖akka，所以我们在工作中是可以使用akka的。当然akka的一些概念比较困难，学习路线比较长，所以想要学会也需要一些时日，必须经过深入学习实战才可。

# 参考资料 #

*  书籍《java高并发程序设计》
 *  [AKKA官方文档][AKKA]

[akka_ akka]: http://blog.csdn.net/liubenlong007/article/details/54603296
[akka_ akka 1]: http://blog.csdn.net/liubenlong007/article/details/54601998
[akka_ Spring_Akka]: http://blog.csdn.net/liubenlong007/article/details/54601448
[akka_ akka 2]: http://blog.csdn.net/liubenlong007/article/details/54584303
[akka_ agent]: http://blog.csdn.net/liubenlong007/article/details/54576611
[akka_ STM]: http://blog.csdn.net/liubenlong007/article/details/54576412
[akka_ Actor_Future-]: http://blog.csdn.net/liubenlong007/article/details/54575312
[akka_ _Procedure]: http://blog.csdn.net/liubenlong007/article/details/54574878
[akka_ _Router]: http://blog.csdn.net/liubenlong007/article/details/54574064
[akka_ inbox]: http://blog.csdn.net/liubenlong007/article/details/54377820
[akka_ actor]: http://blog.csdn.net/liubenlong007/article/details/54093889
[akka_]: http://blog.csdn.net/liubenlong007/article/details/54093794
[akka_HelloWord]: http://blog.csdn.net/liubenlong007/article/details/53782971
[akka]: http://blog.csdn.net/liubenlong007/article/details/53782966
[Akka]: http://blog.csdn.net/beliefer/article/details/53893929
[AKKA]: http://akka.io/docs/