我正在尝试使用Atomix和Netty在Java中的两个进程之间发送消息。
我有一个名为Starter
的程序,该程序负责将消息发送到所有正在等待该消息的正在运行的进程。
这是入门:
Address[] network = {
Address.from("localhost:23450"),
Address.from("localhost:23451"),
Address.from("localhost:23452"),
Address.from("localhost:23453")
};
Serializer s = new SerializerBuilder().build();
ExecutorService es = Executors.newSingleThreadExecutor();
ManagedMessagingService ms = NettyMessagingService.builder()
.withAddress(Address.from("localhost:23459"))
.build();
ms.registerHandler("start", (o,m)->{
System.out.println("Hello "+s.decode(m)+" from "+o);
}, es);
ms.sendAsync(Address.from("localhost:23459"), "start", s.encode("start"));
for (int i = 0; i < network.length; i++) {
ms.sendAsync(network[i], "start", s.encode("start"));
}
这是过程:
Address[] network = {
Address.from("localhost:23450"),
Address.from("localhost:23451"),
Address.from("localhost:23452"),
Address.from("localhost:23453")
};
int id = Integer.parseInt(args[0]);
Serializer s = new SerializerBuilder().addType(Msg.class).build();
ManagedMessagingService ms = NettyMessagingService.builder()
.withAddress(network[id])
.build();
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es2 = Executors.newSingleThreadExecutor();
ms.registerHandler("leader", (o,m) -> {
int i = s.decode(m);
if(leader < i){
leader = i;
System.out.println("Leader updated to: " + i);
}
else{
System.out.println("Received: " + i + ". Not updated.");
}
}, es);
ms.registerHandler("start", (o,m) -> {
for(int i=0; i < network.length ; i++ ){
ms.sendAsync(network[i],"leader", s.encode(id));
}
System.out.println("Started");
},es2);
每个正在运行的进程都有一个ID(作为参数传递)。地址是已知的。
我有两个问题。
第一个是应该等待的进程正在终止。
第二个,
Starter
无法将消息发送到其他进程。我得到:Exception in thread "main" java.lang.NullPointerException: group
我做错了什么?是否应该改变解决问题的方法?
最佳答案
我解决了在两个文件中添加ms.start();
的问题。
最终结果是:
ManagedMessagingService ms = NettyMessagingService.builder()
.withAddress(network[id])
.build();
//(...)
ms.start();
ms.registerHandler("leader", (o,m) -> {
//...
});