我正在尝试使用AtomixNetty在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) -> {
    //...
});

10-04 19:20