Akka与Spring整合应用

發表於 | 分類於 | | 閱讀次數:

本篇集合Akka应用-分布式计算应用展示在Spring项目中做分布式计算.

工程结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
[root@localhost spring-akka ]$ tree akka-task
akka-task
├── pom.xml
├── src
│   └── main
│       ├── java
│       │   └── com
│       │       └── elonsu
│       │           └── cluster
│       │               └── akka
│       │                   ├── Application.java
│       │                   └── spring
│       │                       ├── actor
│       │                       │   ├── Message.java
│       │                       │   ├── RouterActor.java
│       │                       │   └── SimpleActor.java
│       │                       ├── config
│       │                       │   ├── ClientConfig.java
│       │                       │   └── ServerConfig.java
│       │                       ├── controller
│       │                       │   └── AkkaController.java
│       │                       └── support
│       │                           ├── SpringActorProducer.java
│       │                           └── SpringExtension.java
│       └── resources
│           ├── akka-client.conf
│           └── akka-server.conf

Maven依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
<properties>
    <akka.version>2.5.9</akka.version>
    <guava.version>18.0</guava.version>
    <lombok.version>1.16.18</lombok.version>
    <fastjson.version>1.2.41</fastjson.version>
</properties>

<dependencies>
    <dependency>
        <groupId>com.typesafe.akka</groupId>
        <artifactId>akka-cluster_2.11</artifactId>
        <version>${akka.version}</version>
    </dependency>
    <dependency>
        <groupId>com.typesafe.akka</groupId>
        <artifactId>akka-actor_2.11</artifactId>
        <version>${akka.version}</version>
    </dependency>
    <dependency>
        <groupId>com.typesafe.akka</groupId>
        <artifactId>akka-remote_2.11</artifactId>
        <version>${akka.version}</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>${fastjson.version}</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>${lombok.version}</version>
    </dependency>
    <dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>${guava.version}</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
        </plugin>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
        </plugin>
    </plugins>
</build>

模块说明

消息相关

  • Message: 示例中客户端和服务端通讯的消息定义

服务端相关

  • SimpleActor: 服务端事件处理类
  • ServerConfig : 服务端实例化配置
  • akka-server.conf : 服务端配置

客户端相关

  • RouterActor: 消息投递转发类, 用于对于服务端节点进行选取和消息投递
  • ClientConfig: 客户端端实例化配置
  • akka-client.conf : 客户端配置

应用启动类

  • Application: SpringBoot应用主类

容器衔接

  • SpringActorProducer & SpringExtension: IndirectActorProducer是Akka提供的Actor生成接口, SpringActorProducer实现该接口, 以ApplicationContextAware为代理生成我们需要的Actor的Props

模块代码

Spring衔接

SpringActorProducer代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class SpringActorProducer implements IndirectActorProducer {

    final ApplicationContext applicationContext;
    final String actorBeanName;
    final Object[] args;

    public SpringActorProducer(ApplicationContext applicationContext, String actorBeanName, Object... args) {
        this.applicationContext = applicationContext;
        this.actorBeanName = actorBeanName;
        this.args = args;
    }

    @Override
    public Actor produce() {
        return (Actor) applicationContext.getBean(actorBeanName, args);
    }

    @Override
    public Class<? extends Actor> actorClass() {
        return (Class<? extends Actor>) applicationContext.getType(actorBeanName);
    }
}

SpringExtension代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public class SpringExtension extends AbstractExtensionId<SpringExtension.SpringExt> {

    public static SpringExtension INSTANCE = new SpringExtension();

    @Override
    public SpringExt createExtension(ExtendedActorSystem system) {
        return new SpringExt();
    }

    public static class SpringExt implements Extension {

        private volatile ApplicationContext applicationContext;

        public void initialize(ApplicationContext applicationContext) {
            this.applicationContext = applicationContext;
        }

        public Props props(String actorBeanName, Object... args) {
            return Props.create(SpringActorProducer.class, applicationContext, actorBeanName, args);
        }
    }
}
`

消息定义

说明: 消息定义必须实现序列化接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public class Message implements Serializable{

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static final class Task implements Serializable{

        private long  taskId;

        private String content;

        @Override
        public String toString() {
            return JSON.toJSONString(this);
        }
    }

    @Data
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public static final class Result implements Serializable {

        private long taskId;

        private String result;

        @Override
        public String toString() {
            return JSON.toJSONString(this);
        }
    }

}

配置文件

服务端配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
akka {
  loglevel = "INFO"

  actor {
    provider = "akka.cluster.ClusterActorRefProvider"
  }

  remote {
    log-remote-lifecycle-events = on
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2555
    }
  }

  cluster {
    roles = ["server"]
    seed-nodes = [
      "akka.tcp://simpleAkkaCluster@127.0.0.1:2555"]

    auto-down-unreachable-after = 1000s
    auto-down-unreachable-after = 1000s
    metrics.enabled = off
    jmx.multi-mbeans-in-same-jvm = on
  }
}

客户端配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
akka {
  loglevel = "INFO"

  actor {
    provider = "akka.cluster.ClusterActorRefProvider"
  }

  remote {
    log-remote-lifecycle-events = on
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2556
    }
  }

  cluster {
    roles = ["router", "client"]
    seed-nodes = [
      "akka.tcp://simpleAkkaCluster@127.0.0.1:2555"]

    auto-down-unreachable-after = 1000s
    auto-down-unreachable-after = 1000s
    metrics.enabled = off
    jmx.multi-mbeans-in-same-jvm = on
  }

}

Actor示例

服务端消息处理Actor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@Component
@Scope("prototype")
public class SimpleActor extends AbstractLoggingActor{

    // 集群监听
    private Cluster cluster = Cluster.get(getContext().system());

    //subscribe to cluster changes
    @Override
    public void preStart() {
        //#subscribe
        cluster.subscribe(getSelf(), ClusterEvent.initialStateAsEvents(),
                ClusterEvent.MemberEvent.class, ClusterEvent.UnreachableMember.class);
        //#subscribe
    }

    //re-subscribe when restart
    @Override
    public void postStop() {
        cluster.unsubscribe(getSelf());
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(Message.Task.class,
                        x-> {
                            log().info("[接收消息]:{}, Actor:{}, Thread:{}", x, getSelf().path(), Thread.currentThread().getName());
                            getSender().tell(new Message.Result(x.getTaskId(), x.getContent() + "处理完成"), self());
                        }
                )
                .match(ClusterEvent.MemberUp.class,
                        x -> log().info("Member is Up: {}", x.member())
                )
                .match(ClusterEvent.CurrentClusterState.class,
                        x -> log().info("Cluster State: {}", x.getMembers())
                )
                .match(ClusterEvent.UnreachableMember.class,
                        x -> log().info("Member detected as unreachable: {}", x.member())
                )
                .match(ClusterEvent.MemberRemoved.class,
                        x -> log().info("Member is Removed: {}", x.member())
                )
                .match(ClusterEvent.MemberEvent.class,
                        x -> log().info("ignore")
                )
                .matchAny(
                        x -> unhandled(x)
                )
                .build();
    }

}

客户端路由分发Actor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
@Component
@Scope("prototype")
public class RouterActor extends AbstractLoggingActor {

    // 服务集合
    private List<ActorRef> servers = Lists.newArrayList();

    // 集群监听
    private Cluster cluster = Cluster.get(context().system());

    private AtomicInteger count = new AtomicInteger(0);

    private String actorPath;


    public RouterActor(String actorPath) {
        this.actorPath = actorPath;
    }

    @Override
    public void preStart() throws Exception {
        cluster.subscribe(self(), ClusterEvent.MemberEvent.class, ClusterEvent.ReachabilityEvent.class);
    }

    @Override
    public void postStop() throws Exception {
        cluster.unsubscribe(self());
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(Message.Task.class,
                        x -> servers.isEmpty(),
                        x -> sender().tell(new Message.Result(-1L, "服务暂不可用"), sender())
                )
                .match(Message.Task.class,
                        x -> {
                            log().info("[Dispatcher][消息转发] 集群节点数量:{}, 待处理任务:{}", servers.size(), x.getTaskId());

                            int index = count.incrementAndGet() % servers.size();
                            for(int i=0; i<servers.size(); i++){
                                log().info("[Dispatcher][服务选取] 选中[" + (i == index) + "] 地址: " + servers.get(i).path());
                            }
                            //System.out.println();
                            servers.get(index).forward(x, getContext());
                        }
                )
                .match(Terminated.class,
                        x -> unRegister(x.getActor())
                )
                .match(ClusterEvent.CurrentClusterState.class, state -> {
                    // 当前节点在刚刚加入集群时，会收到CurrentClusterState消息，从中可以解析出集群中的所有前端节点
                    servers.clear();
                    for (Member member : state.getMembers()) {
                        if (member.status().equals(MemberStatus.up())) {
                            register(member);
                        }
                    }
                })
                .match(ClusterEvent.MemberUp.class,
                        x -> register(x.member())
                )
                .match(ClusterEvent.MemberEvent.class,
                        x -> unRegister(x.member())
                )
                .match(ClusterEvent.UnreachableMember.class,
                        x -> unRegister(x.member())
                )
                .match(ClusterEvent.ReachableMember.class,
                        x -> unRegister(x.member())
                ).build();
    }


    public void register(Member member){
        if(null != member && member.hasRole("server")){
            register(getContext().actorFor(member.address() + "/user/" + this.actorPath));
        }
    }

    public void unRegister(Member member){
        if(null != member && member.hasRole("server")){
            unRegister(getContext().actorFor(member.address() + "/user/" + this.actorPath));
        }
    }

    public void register(ActorRef actorRef){
        if(null != actorRef){
            servers.add(actorRef);
        }
    }

    public void unRegister(ActorRef actorRef){
        if(null != actorRef){
            servers.remove(actorRef);
        }
    }

    public void setActorPath(String actorPath) {
        this.actorPath = actorPath;
    }
}

集群实例化

服务端注解实例化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@Slf4j
@Configuration
public class ServerConfig {

    @Autowired
    private ApplicationContext applicationContext;

    private Config config;

    @PostConstruct
    public void init(){
        load("akka-server.conf");
    }

    private void load(String resource){
        config = ConfigFactory.load(this.getClass().getClassLoader(), resource);
        List<String> roles = config.getStringList("akka.cluster.roles");
        log.info("[Cluster Client][Role] :{}", roles);
    }


    @Bean("serverSystem")
    public ActorSystem actorSystem() {
        ActorSystem system = ActorSystem.create("simpleAkkaCluster", config);
        // initialize the application context in the Akka Spring Extension
        SpringExtension.INSTANCE.get(system).initialize(applicationContext);
        log.info("[Cluster Client] [System] {}", system.name());
        return system;
    }

    @Bean("serverActor")
    public ActorRef actorRef(){
        ActorSystem system = actorSystem();
        ActorRef ebIndexActor = system.actorOf(
                SpringExtension.INSTANCE.get(system).props("simpleActor"),
                "SimpleEcho");
        log.info("[Cluster Server Node] [Loaded] {} ", ebIndexActor.path());
        return ebIndexActor;
    }
}

客户端注解实例化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@Slf4j
@Configuration
public class ClientConfig {

    @Autowired
    private ApplicationContext applicationContext;

    private Config config;

    @PostConstruct
    public void init(){
        load("akka-client.conf");
    }

    private void load(String resource){
        config = ConfigFactory.load(this.getClass().getClassLoader(), resource);
        List<String> roles = config.getStringList("akka.cluster.roles");
        log.info("[Cluster Client][Role] :{}", roles);
    }


    @Bean("clientSystem")
    public ActorSystem actorSystem() {
        ActorSystem system = ActorSystem.create("simpleAkkaCluster", config);
        // initialize the application context in the Akka Spring Extension
        SpringExtension.INSTANCE.get(system).initialize(applicationContext);
        log.info("[Cluster Client] [System] {}", system.name());
        return system;
    }

    @Bean("clientActor")
    public ActorRef actorRef(){
        ActorSystem system = actorSystem();
        ActorRef clientActor = system.actorOf(SpringExtension.INSTANCE.get(system).props("routerActor", "SimpleEcho"), "clientActor");
        log.info("[Cluster Client Node] [Loaded] {} ", clientActor.path());
        return clientActor;
    }
}

Rest测试类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Slf4j
@RestController
@RequestMapping("/api/akka")
public class AkkaController {

    @Resource(name = "clientActor")
    private ActorRef clientActor;

    /**
     * 和远程端进行数据交互并获取处理结果
     * @throws Exception
     */
    @GetMapping(value = "/cluster/event", produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
    public ResponseEntity clusterEvent() throws Exception {
        log.info("[Client] path:{}", clientActor.path());

        List<Future<Object>> futures = Lists.newArrayList();
        Timeout timeout = new Timeout(Duration.create(50, TimeUnit.SECONDS));
        for(int i=0; i<5; i++){
            String text = "集群消息" + i + UUID.randomUUID().toString();
            Future<Object> future = Patterns.ask(clientActor, new Message.Task(100L +i, text), timeout);
            futures.add(future);
        }

        for(Future<Object> future: futures){
            Message.Result result = (Message.Result) Await.result(future, timeout.duration());
            System.out.println("服务端返回:" + result);
        }

        return ResponseEntity.ok("test");
    }
}

SpringBoot主类

1
2
3
4
5
6
7
8
9
@Slf4j
@SpringBootApplication
public class Application {

    public static void main(String[] args) throws Exception{
        SpringApplication.run(Application.class, args);
        log.info("[启动成功] 应用名称: {}", "Akka Spring Boot");
    }
}

访问测试

启动SpringBoot应用

1
2
3
4
5
6
7
8
9
10
11
12
2018-02-05 17:54:13.586  INFO 6523 --- [           main] s.b.c.e.t.TomcatEmbeddedServletContainer : Tomcat started on port(s): 8080 (http)
2018-02-05 17:54:13.594  INFO 6523 --- [           main] com.elonsu.cluster.akka.Application      : Started Application in 4.948 seconds (JVM running for 6.118)
2018-02-05 17:54:13.594  INFO 6523 --- [           main] com.elonsu.cluster.akka.Application      : [启动成功] 应用名称: Akka Spring Boot
[INFO] [02/05/2018 17:54:17.602] [simpleAkkaCluster-akka.actor.default-dispatcher-3] [akka://simpleAkkaCluster/deadLetters] Message [akka.cluster.InternalClusterAction$InitJoin$] from Actor[akka://simpleAkkaCluster/system/cluster/core/daemon/joinSeedNodeProcess-1#-1342745044] to Actor[akka://simpleAkkaCluster/deadLetters] was not delivered. [2] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[WARN] [02/05/2018 17:54:22.622] [simpleAkkaCluster-akka.actor.default-dispatcher-15] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/system/cluster/core/daemon/joinSeedNodeProcess-1] Couldn't join seed nodes after [2] attempts, will try again. seed-nodes=[akka.tcp://simpleAkkaCluster@127.0.0.1:2555]
[INFO] [02/05/2018 17:54:22.698] [simpleAkkaCluster-akka.actor.default-dispatcher-3] [akka.cluster.Cluster(akka://simpleAkkaCluster)] Cluster Node [akka.tcp://simpleAkkaCluster@127.0.0.1:2555] - Received InitJoin message from [Actor[akka.tcp://simpleAkkaCluster@127.0.0.1:2556/system/cluster/core/daemon/joinSeedNodeProcess-1#-1342745044]] to [akka.tcp://simpleAkkaCluster@127.0.0.1:2555]
[INFO] [02/05/2018 17:54:22.698] [simpleAkkaCluster-akka.actor.default-dispatcher-3] [akka.cluster.Cluster(akka://simpleAkkaCluster)] Cluster Node [akka.tcp://simpleAkkaCluster@127.0.0.1:2555] - Sending InitJoinAck message from node [akka.tcp://simpleAkkaCluster@127.0.0.1:2555] to [Actor[akka.tcp://simpleAkkaCluster@127.0.0.1:2556/system/cluster/core/daemon/joinSeedNodeProcess-1#-1342745044]]
[INFO] [02/05/2018 17:54:22.721] [simpleAkkaCluster-akka.actor.default-dispatcher-19] [akka.cluster.Cluster(akka://simpleAkkaCluster)] Cluster Node [akka.tcp://simpleAkkaCluster@127.0.0.1:2555] - Node [akka.tcp://simpleAkkaCluster@127.0.0.1:2556] is JOINING, roles [router, client, dc-default]
[INFO] [02/05/2018 17:54:22.723] [simpleAkkaCluster-akka.actor.default-dispatcher-2] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] ignore
[INFO] [02/05/2018 17:54:22.726] [simpleAkkaCluster-akka.actor.default-dispatcher-19] [akka.cluster.Cluster(akka://simpleAkkaCluster)] Cluster Node [akka.tcp://simpleAkkaCluster@127.0.0.1:2555] - Leader is moving node [akka.tcp://simpleAkkaCluster@127.0.0.1:2556] to [Up]
[INFO] [02/05/2018 17:54:22.726] [simpleAkkaCluster-akka.actor.default-dispatcher-24] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] Member is Up: Member(address = akka.tcp://simpleAkkaCluster@127.0.0.1:2556, status = Up)
[INFO] [02/05/2018 17:54:22.828] [simpleAkkaCluster-akka.actor.default-dispatcher-15] [akka.cluster.Cluster(akka://simpleAkkaCluster)] Cluster Node [akka.tcp://simpleAkkaCluster@127.0.0.1:2556] - Welcome from [akka.tcp://simpleAkkaCluster@127.0.0.1:2555

访问接口

1
2
[root@localhost spring-akka ]$ curl http://localhost:8080/api/akka/cluster/event
test

交互日志

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
2018-02-05 17:55:37.462  INFO 6523 --- [nio-8080-exec-9] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring FrameworkServlet 'dispatcherServlet'
2018-02-05 17:55:37.462  INFO 6523 --- [nio-8080-exec-9] o.s.web.servlet.DispatcherServlet        : FrameworkServlet 'dispatcherServlet': initialization started
2018-02-05 17:55:37.490  INFO 6523 --- [nio-8080-exec-9] o.s.web.servlet.DispatcherServlet        : FrameworkServlet 'dispatcherServlet': initialization completed in 28 ms
2018-02-05 17:55:37.523  INFO 6523 --- [nio-8080-exec-9] c.e.c.a.s.controller.AkkaController      : [Client] path:akka://simpleAkkaCluster/user/clientActor
[INFO] [02/05/2018 17:55:37.526] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:100
[INFO] [02/05/2018 17:55:37.526] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/05/2018 17:55:37.527] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:101
[INFO] [02/05/2018 17:55:37.527] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/05/2018 17:55:37.527] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:102
[INFO] [02/05/2018 17:55:37.527] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/05/2018 17:55:37.527] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:103
[INFO] [02/05/2018 17:55:37.527] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/05/2018 17:55:37.527] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:104
[INFO] [02/05/2018 17:55:37.527] [simpleAkkaCluster-akka.actor.default-dispatcher-4] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[WARN] [SECURITY][02/05/2018 17:55:37.528] [simpleAkkaCluster-akka.remote.default-remote-dispatcher-18] [akka.serialization.Serialization(akka://simpleAkkaCluster)] Using the default Java serializer for class [com.elonsu.cluster.akka.spring.actor.Message$Task] which is not recommended because of performance implications. Use another serializer or disable this warning using the setting 'akka.actor.warn-about-java-serializer-usage'
[INFO] [02/05/2018 17:55:37.616] [simpleAkkaCluster-akka.actor.default-dispatcher-16] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息0526add9d-29ba-409f-ae74-cc7b61606613","taskId":100}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-16
[INFO] [02/05/2018 17:55:37.616] [simpleAkkaCluster-akka.actor.default-dispatcher-16] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息16d03f841-67d4-4169-b747-dc0a8f2e02dd","taskId":101}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-16
[INFO] [02/05/2018 17:55:37.616] [simpleAkkaCluster-akka.actor.default-dispatcher-16] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息25c2c2579-71c5-45a8-97e6-434a95a186df","taskId":102}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-16
[INFO] [02/05/2018 17:55:37.617] [simpleAkkaCluster-akka.actor.default-dispatcher-16] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息3f4ad2fe3-d624-416d-a01a-e346914fcbbb","taskId":103}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-16
[INFO] [02/05/2018 17:55:37.617] [simpleAkkaCluster-akka.actor.default-dispatcher-16] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息421b5b619-345d-4eb0-8682-df8955ea70df","taskId":104}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-16
[WARN] [SECURITY][02/05/2018 17:55:37.617] [simpleAkkaCluster-akka.remote.default-remote-dispatcher-21] [akka.serialization.Serialization(akka://simpleAkkaCluster)] Using the default Java serializer for class [com.elonsu.cluster.akka.spring.actor.Message$Result] which is not recommended because of performance implications. Use another serializer or disable this warning using the setting 'akka.actor.warn-about-java-serializer-usage'
服务端返回:{"result":"集群消息0526add9d-29ba-409f-ae74-cc7b61606613处理完成","taskId":100}
服务端返回:{"result":"集群消息16d03f841-67d4-4169-b747-dc0a8f2e02dd处理完成","taskId":101}
服务端返回:{"result":"集群消息25c2c2579-71c5-45a8-97e6-434a95a186df处理完成","taskId":102}
服务端返回:{"result":"集群消息3f4ad2fe3-d624-416d-a01a-e346914fcbbb处理完成","taskId":103}
服务端返回:{"result":"集群消息421b5b619-345d-4eb0-8682-df8955ea70df处理完成","taskId":104}

集群部署

上面示例演示的在同一台机器上的不同端口进行通讯. 接下来演示在多台机器上的不同配置

服务端配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
akka {
  loglevel = "INFO"

  actor {
    provider = "akka.cluster.ClusterActorRefProvider"
  }

  remote {
    log-remote-lifecycle-events = on
    netty.tcp {
      port = 2555
    }
  }

  cluster {
    roles = ["server"]
    seed-nodes = [
      "akka.tcp://simpleAkkaCluster@192.168.0.101:2555",
      "akka.tcp://simpleAkkaCluster@192.168.0.102:2555",
      "akka.tcp://simpleAkkaCluster@192.168.0.103:2555"]

    auto-down-unreachable-after = 1000s
    auto-down-unreachable-after = 1000s
    metrics.enabled = off
    jmx.multi-mbeans-in-same-jvm = on
  }
}

客户端配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
akka {
  loglevel = "INFO"

  actor {
    provider = "akka.cluster.ClusterActorRefProvider"
  }

  remote {
    log-remote-lifecycle-events = on
    netty.tcp {
      port = 2556
    }
  }

  cluster {
    roles = ["router", "client"]
    seed-nodes = [
      "akka.tcp://simpleAkkaCluster@192.168.0.101:2555",
      "akka.tcp://simpleAkkaCluster@192.168.0.102:2555",
      "akka.tcp://simpleAkkaCluster@192.168.0.103:2555"]

    auto-down-unreachable-after = 1000s
    auto-down-unreachable-after = 1000s
    metrics.enabled = off
    jmx.multi-mbeans-in-same-jvm = on
  }

}

相对于单节点配置,不同的是示例中服务节点都配置为不同IP, 同时摘掉了绑定的akka.remote.netty.tcp.hostname, 让Akka自动获取实例的通讯IP进行绑定.

配置Dispatcher

客户端配置重定义默认的default-dispatcher

配置thread-pool默认默认

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
akka {
  loglevel = "INFO"

  actor {
    provider = "akka.cluster.ClusterActorRefProvider"

    # default dispatcher
    default-dispatcher {
      type = "Dispatcher"
      # What kind of ExecutionService to use
      executor = "thread-pool-executor"
      # Configuration for the thread pool
      thread-pool-executor {
        # minimum number of threads to cap factor-based core number to
        core-pool-size-min = 3
        # No of core threads ... ceil(available processors * factor)
        core-pool-size-factor = 2.0
        # maximum number of threads to cap factor-based number to
        core-pool-size-max = 10
      }
      # Throughput defines the maximum number of messages to be
      # processed per actor before the thread jumps to the next actor.
      # Set to 1 for as fair as possible.
      throughput = 20
    }

  }

  remote {
    log-remote-lifecycle-events = on
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2556
    }
  }

  cluster {
    roles = ["router", "client"]
    seed-nodes = [
      "akka.tcp://simpleAkkaCluster@127.0.0.1:2555"]

    auto-down-unreachable-after = 1000s
    auto-down-unreachable-after = 1000s
    metrics.enabled = off
    jmx.multi-mbeans-in-same-jvm = on
  }

}

执行日志

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
2018-02-06 11:49:28.222  INFO 2218 --- [nio-8080-exec-7] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring FrameworkServlet 'dispatcherServlet'
2018-02-06 11:49:28.223  INFO 2218 --- [nio-8080-exec-7] o.s.web.servlet.DispatcherServlet        : FrameworkServlet 'dispatcherServlet': initialization started
2018-02-06 11:49:28.238  INFO 2218 --- [nio-8080-exec-7] o.s.web.servlet.DispatcherServlet        : FrameworkServlet 'dispatcherServlet': initialization completed in 15 ms
2018-02-06 11:49:28.260  INFO 2218 --- [nio-8080-exec-7] c.e.c.a.s.controller.AkkaController      : [Client] path:akka://simpleAkkaCluster/user/clientActor
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:100
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:101
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:102
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:103
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:104
[INFO] [02/06/2018 11:49:28.262] [simpleAkkaCluster-akka.actor.default-dispatcher-9] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[WARN] [SECURITY][02/06/2018 11:49:28.263] [simpleAkkaCluster-akka.remote.default-remote-dispatcher-11] [akka.serialization.Serialization(akka://simpleAkkaCluster)] Using the default Java serializer for class [com.elonsu.cluster.akka.spring.actor.Message$Task] which is not recommended because of performance implications. Use another serializer or disable this warning using the setting 'akka.actor.warn-about-java-serializer-usage'
[INFO] [02/06/2018 11:49:28.329] [simpleAkkaCluster-akka.actor.default-dispatcher-22] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息06ab614d0-f50d-4c6c-b91d-7b5af1e98187","taskId":100}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-22
[INFO] [02/06/2018 11:49:28.330] [simpleAkkaCluster-akka.actor.default-dispatcher-22] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息1057874e5-1219-4bea-81f5-5612707bf83d","taskId":101}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-22
[INFO] [02/06/2018 11:49:28.330] [simpleAkkaCluster-akka.actor.default-dispatcher-22] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息2467e56fa-e2dd-493a-a583-cf0496bbb96f","taskId":102}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-22
[INFO] [02/06/2018 11:49:28.330] [simpleAkkaCluster-akka.actor.default-dispatcher-22] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息3ccad0171-99b4-4384-b836-72115304f77a","taskId":103}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-22
[INFO] [02/06/2018 11:49:28.330] [simpleAkkaCluster-akka.actor.default-dispatcher-22] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息427786c84-4971-4f4d-9c52-a4975b9548bf","taskId":104}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-22
[WARN] [SECURITY][02/06/2018 11:49:28.330] [simpleAkkaCluster-akka.remote.default-remote-dispatcher-5] [akka.serialization.Serialization(akka://simpleAkkaCluster)] Using the default Java serializer for class [com.elonsu.cluster.akka.spring.actor.Message$Result] which is not recommended because of performance implications. Use another serializer or disable this warning using the setting 'akka.actor.warn-about-java-serializer-usage'
服务端返回:{"result":"集群消息06ab614d0-f50d-4c6c-b91d-7b5af1e98187处理完成","taskId":100}
服务端返回:{"result":"集群消息1057874e5-1219-4bea-81f5-5612707bf83d处理完成","taskId":101}
服务端返回:{"result":"集群消息2467e56fa-e2dd-493a-a583-cf0496bbb96f处理完成","taskId":102}
服务端返回:{"result":"集群消息3ccad0171-99b4-4384-b836-72115304f77a处理完成","taskId":103}
服务端返回:{"result":"集群消息427786c84-4971-4f4d-9c52-a4975b9548bf处理完成","taskId":104}

配置fork-join默认默认

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
akka {
  loglevel = "INFO"

  actor {
    provider = "akka.cluster.ClusterActorRefProvider"

    # self defined dispatcher
    default-dispatcher {
      type = "Dispatcher"
      # What kind of ExecutionService to use
      executor = "fork-join-executor"
      # Configuration for the fork join pool
      fork-join-executor {
        # Min number of threads to cap factor-based parallelism number to
        parallelism-min = 2
        # Parallelism (threads) ... ceil(available processors * factor)
        parallelism-factor = 2.0
        # Max number of threads to cap factor-based parallelism number to
        parallelism-max = 10
      }
      # Throughput defines the maximum number of messages to be
      # processed per actor before the thread jumps to the next actor.
      # Set to 1 for as fair as possible.
      throughput = 20
    }

  }

  remote {
    log-remote-lifecycle-events = on
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2556
    }
  }

  cluster {
    roles = ["router", "client"]
    seed-nodes = [
      "akka.tcp://simpleAkkaCluster@127.0.0.1:2555"]

    auto-down-unreachable-after = 1000s
    auto-down-unreachable-after = 1000s
    metrics.enabled = off
    jmx.multi-mbeans-in-same-jvm = on
  }

}

客户端配置自定义Dispatcher

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
akka {
  loglevel = "INFO"

  actor {
    provider = "akka.cluster.ClusterActorRefProvider"

    # self defined dispatcher
    thread-pool-dispatcher {
      type = "Dispatcher"
      # What kind of ExecutionService to use
      executor = "thread-pool-executor"
      # Configuration for the thread pool
      thread-pool-executor {
        # minimum number of threads to cap factor-based core number to
        core-pool-size-min = 3
        # No of core threads ... ceil(available processors * factor)
        core-pool-size-factor = 2.0
        # maximum number of threads to cap factor-based number to
        core-pool-size-max = 10
      }
      # Throughput defines the maximum number of messages to be
      # processed per actor before the thread jumps to the next actor.
      # Set to 1 for as fair as possible.
      throughput = 20
    }


    deployment {
      /clientActor {

        # 使用自定义的Dispatcher
        dispatcher = "akka.actor.thread-pool-dispatcher"
      }
    }

  }

  remote {
    log-remote-lifecycle-events = on
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2556
    }
  }

  cluster {
    roles = ["router", "client"]
    seed-nodes = [
      "akka.tcp://simpleAkkaCluster@127.0.0.1:2555"]

    auto-down-unreachable-after = 1000s
    auto-down-unreachable-after = 1000s
    metrics.enabled = off
    jmx.multi-mbeans-in-same-jvm = on
  }

}

执行输出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
2018-02-06 11:55:59.839  INFO 2301 --- [nio-8080-exec-3] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring FrameworkServlet 'dispatcherServlet'
2018-02-06 11:55:59.839  INFO 2301 --- [nio-8080-exec-3] o.s.web.servlet.DispatcherServlet        : FrameworkServlet 'dispatcherServlet': initialization started
2018-02-06 11:55:59.852  INFO 2301 --- [nio-8080-exec-3] o.s.web.servlet.DispatcherServlet        : FrameworkServlet 'dispatcherServlet': initialization completed in 13 ms
2018-02-06 11:55:59.872  INFO 2301 --- [nio-8080-exec-3] c.e.c.a.s.controller.AkkaController      : [Client] path:akka://simpleAkkaCluster/user/clientActor
[INFO] [02/06/2018 11:55:59.873] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:100
[INFO] [02/06/2018 11:55:59.873] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/06/2018 11:55:59.874] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:101
[INFO] [02/06/2018 11:55:59.874] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/06/2018 11:55:59.874] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:102
[INFO] [02/06/2018 11:55:59.874] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/06/2018 11:55:59.874] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:103
[INFO] [02/06/2018 11:55:59.874] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[INFO] [02/06/2018 11:55:59.874] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][消息转发] 集群节点数量:1, 待处理任务:104
[INFO] [02/06/2018 11:55:59.874] [simpleAkkaCluster-akka.actor.thread-pool-dispatcher-26] [akka.tcp://simpleAkkaCluster@127.0.0.1:2556/user/clientActor] [Dispatcher][服务选取] 选中[true] 地址: akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho
[WARN] [SECURITY][02/06/2018 11:55:59.874] [simpleAkkaCluster-akka.remote.default-remote-dispatcher-5] [akka.serialization.Serialization(akka://simpleAkkaCluster)] Using the default Java serializer for class [com.elonsu.cluster.akka.spring.actor.Message$Task] which is not recommended because of performance implications. Use another serializer or disable this warning using the setting 'akka.actor.warn-about-java-serializer-usage'
[INFO] [02/06/2018 11:55:59.926] [simpleAkkaCluster-akka.actor.default-dispatcher-2] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息0a860813a-d7c3-4a2a-b93a-35b034016731","taskId":100}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-2
[INFO] [02/06/2018 11:55:59.926] [simpleAkkaCluster-akka.actor.default-dispatcher-2] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息1c3148d4d-de3f-4483-b668-19b608b8d83d","taskId":101}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-2
[INFO] [02/06/2018 11:55:59.926] [simpleAkkaCluster-akka.actor.default-dispatcher-2] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息26707b125-a867-484a-9c5c-21a085d31b89","taskId":102}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-2
[INFO] [02/06/2018 11:55:59.926] [simpleAkkaCluster-akka.actor.default-dispatcher-2] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息334e7fdd0-c3cc-4192-8ee5-89f28c1dbe28","taskId":103}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-2
[INFO] [02/06/2018 11:55:59.927] [simpleAkkaCluster-akka.actor.default-dispatcher-2] [akka.tcp://simpleAkkaCluster@127.0.0.1:2555/user/SimpleEcho] [接收消息]:{"content":"集群消息450ee1130-33fb-4f23-9372-34cff2d52488","taskId":104}, Actor:akka://simpleAkkaCluster/user/SimpleEcho, Thread:simpleAkkaCluster-akka.actor.default-dispatcher-2
[WARN] [SECURITY][02/06/2018 11:55:59.926] [simpleAkkaCluster-akka.remote.default-remote-dispatcher-16] [akka.serialization.Serialization(akka://simpleAkkaCluster)] Using the default Java serializer for class [com.elonsu.cluster.akka.spring.actor.Message$Result] which is not recommended because of performance implications. Use another serializer or disable this warning using the setting 'akka.actor.warn-about-java-serializer-usage'
服务端返回:{"result":"集群消息0a860813a-d7c3-4a2a-b93a-35b034016731处理完成","taskId":100}
服务端返回:{"result":"集群消息1c3148d4d-de3f-4483-b668-19b608b8d83d处理完成","taskId":101}
服务端返回:{"result":"集群消息26707b125-a867-484a-9c5c-21a085d31b89处理完成","taskId":102}
服务端返回:{"result":"集群消息334e7fdd0-c3cc-4192-8ee5-89f28c1dbe28处理完成","taskId":103}
服务端返回:{"result":"集群消息450ee1130-33fb-4f23-9372-34cff2d52488处理完成","taskId":104}

相关文档