0

我正在尝试创建一个示例，该示例创建一个能够处理订单的集群和一个发送订单的客户端。

为此，我使用 ClusterClientReceptionist 和 ClusterClient 在客户端和集群网络之间进行通信。

集群节点代码：

namespace ClusterNode
{
    class Program
    {
        static void Main(string[] args)
        {
            bool isFirst = false;
            using (var mutex = new Mutex(false, "5D14E338-616C-4637-93E7-E59D152DC005", out isFirst))
            {
                using (var system = ActorSystem.Create(nameof(Order), CreateConfig(isFirst)))
                {
                    var receptionistConfig = ClusterClientReceptionist.DefaultConfig();
                    system.Settings.InjectTopLevelFallback(receptionistConfig);

                    ClusterRouterPoolSettings settings = new ClusterRouterPoolSettings(int.MaxValue, 1, true);
                    ClusterRouterPool pool = new ClusterRouterPool(new Akka.Routing.RoundRobinPool(1), settings);

                    var aref = system.ActorOf(pool.Props(Props.Create<OrderProcessor>()), nameof(OrderProcessor));

                    var receptionist = ClusterClientReceptionist.Get(system);
                    receptionist.RegisterService(aref);

                    system.WhenTerminated.Wait();
                }
            }
        }

        private static Config CreateConfig(bool isFirst)
        {
            var configString = "akka { actor.provider = \"Akka.Cluster.ClusterActorRefProvider, Akka.Cluster\"\n";

            var port = (isFirst) ? 8082 : 0;

            configString += "remote { helios.tcp { hostname = localhost, port = " + port + " } }\n";
            configString += "cluster {\n seed-nodes = [\"akka.tcp://" + nameof(Order) + "@localhost:8082\"] \n";
            configString += "}\n";
            configString += "}";
            var config = ConfigurationFactory.ParseString(configString);
            return config;
        }
    }
}

客户端代码：

namespace Client
{
    class Program
    {
        static void Main(string[] args)
        {
            using (var system = ActorSystem.Create(nameof(Client), CreateConfig()))
            {
                system.Settings.InjectTopLevelFallback(ClusterClientReceptionist.DefaultConfig());
                var client = system.ActorOf(ClusterClient.Props(ClusterClientSettings.Create(system)));

                bool quit = false;
                while (!quit)
                {
                    var str = Console.ReadLine();

                    if (string.Equals(str, "QUIT", StringComparison.InvariantCultureIgnoreCase))
                    {
                        quit = true;
                        continue;
                    }

                    Order order = new Order() { Msg = str };

                    var respTask = client.Ask<Response>(new ClusterClient.Send("/user/OrderProcessor", order));
                    respTask.Wait();

                    Console.WriteLine(respTask.Result.ResponseMsg);
                }
                system.Terminate();
                system.WhenTerminated.Wait();
            }
        }

        private static Config CreateConfig()
        {
            var configString = "akka { actor.provider = \"Akka.Remote.RemoteActorRefProvider, Akka.Remote\"\n";
            configString += "remote { helios.tcp { hostname = localhost, port = 0 } }\n";
            configString += "cluster {\n seed-nodes = [\"akka.tcp://" + nameof(Order) + "@localhost:8082\"], ";
            configString += " client { initial-contacts : [ \"akka.tcp://" + nameof(Order) + "@localhost:8082/system/receptionist\" ] }";
            configString += " }\n";
            configString += "}";
            var config = ConfigurationFactory.ParseString(configString);
            return config;
        }
    }
}

当我创建一个集群节点和一个客户端时，示例工作。如果我尝试创建多个集群节点，它们相互连接然后直接解除关联并且集群崩溃。

如果我评论以下两条线，集群可以工作，但接待员/客户机制不起作用。

var receptionist = ClusterClientReceptionist.Get(system);
receptionist.RegisterService(aref);

他们的配置是否缺失？

4

1 回答 1

1

我今天遇到了同样的问题，并设法通过更改为 Wire 序列化而不是默认的 Json 来解决它。

我必须从 NuGet 添加 Akka.Serialization.Wire 并更新配置中的序列化绑定，您可以在集群工具示例中看到它正在使用

于 2017-08-02T15:34:45.867 回答