1

我正在创建一个项目，此时有一个演员（用户）通过一致的哈希组路由器调用另一个演员（音乐会）。一切正常，但我的问题是音乐会演员我无法回答询问消息。不知何故，消息丢失了，客户端什么也没有发生。我已经尝试了一切都没有运气：

  • Sender.Tell <-- 创建一个时间？发件人
  • 通过消息中的引用传递用户 IActorRef 并使用它。

这是完整的代码：https ://github.com/pablocastilla/AkkaConcert

主要细节如下：

用户演员：

        protected IActorRef concertRouter;


    public User(IActorRef concertRouter, int eventId)
    {
        this.concertRouter = concertRouter;
        this.eventId = eventId;

        JobStarter = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(TimeSpan.FromMilliseconds(20),
                     TimeSpan.FromMilliseconds(1000), Self, new AttemptToStartJob(), Self);

        Receive<AttemptToStartJob>(start =>
        {
            var self = Self;

            concertRouter.Ask<Routees>(new GetRoutees()).ContinueWith(tr =>
            {
                if (tr.Result.Members.Count() > 0)
                {
                    var m = new GetAvailableSeats() { User = self, ConcertId = eventId };

                    self.Tell(m);

                    //                        JobStarter.Cancel();                      

                }

            }, TaskContinuationOptions.ExecuteSynchronously);
        });

        Receive<GetAvailableSeats>(rs =>
        {
            rs.User = Self;

            //get free seats
            concertRouter.Ask(rs).ContinueWith(t=>
            {
                Console.WriteLine("response received!!");

            }
            );
        });

客户端HOCON配置：

  <akka>
<hocon>
  <![CDATA[
                akka {
          actor {

                        provider = "Akka.Cluster.ClusterActorRefProvider, Akka.Cluster"
                        deployment {


                            /eventpool {
                                router = consistent-hashing-group
              routees.paths = ["/user/HugeEvent"]                                   
                                virtual-nodes-factor = 8
                                cluster {
                                        enabled = on
                                        max-nr-of-instances-per-node = 2
                                        allow-local-routees = off
                                        use-role = cluster
                                }
                            }                
                        }
                    }


                    remote {
                        log-remote-lifecycle-events = DEBUG

                        helios.tcp {
                            transport-class = "Akka.Remote.Transport.Helios.HeliosTcpTransport, Akka.Remote"
                            applied-adapters = []
                            transport-protocol = tcp
                            #will be populated with a dynamic host-name at runtime if left uncommented
                            #public-hostname = "POPULATE STATIC IP HERE"
                            hostname = "127.0.0.1"
                            port = 0
                        }
                    }            

                    cluster {
                        #will inject this node as a self-seed node at run-time
                        seed-nodes = ["akka.tcp://akkaconcert@127.0.0.1:8080"] #manually populate other seed nodes here, i.e. "akka.tcp://lighthouse@127.0.0.1:4053", "akka.tcp://lighthouse@127.0.0.1:4044"
                        roles = [client]
                        auto-down-unreachable-after = 60s
                    }
                }
        ]]>
</hocon>

在后端：

演员已创建

   private ActorSystem actorSystem;

    private IActorRef event1;

    public bool Start(HostControl hostControl)
    {
        actorSystem = ActorSystem.Create("akkaconcert");

        SqlServerPersistence.Init(actorSystem);

        event1 = actorSystem.ActorOf(
                Props.Create(() => new Concert(1,100000)), "HugeEvent");

        return true;
    }

音乐会演员消息处理

        private void ReadyCommands()
    {
        Command<GetAvailableSeats>(message => GetFreeSeatsHandler(message));

        Command<ReserveSeats>(message => ReserveSeatsHandler(message));

        Command<BuySeats>(message => Persist(message, BuySeatsHandler));


    }

    private bool GetFreeSeatsHandler(GetAvailableSeats message)
    {          

        var freeSeats = seats.Where(s => s.Value.State == Actors.Seat.SeatState.Free).Select(s2 => s2.Value).ToList();

        //1. Trying passing the user actor
        //message.User.Tell(new GetFreeSeatsResponse() { FreeSeats = freeSeats }, Context.Self);

        //2. Trying with the sender
        Context.Sender.Tell(new GetAvailableSeatsResponse() { FreeSeats = freeSeats }, Context.Self);


        printMessagesPerSecond(messagesReceived++);
        printfreeSeats(freeSeats);

        return true;
    }

后端的 HOCON 配置：

  <akka>
<hocon>
  <![CDATA[
                akka {
                    actor {
                        provider = "Akka.Cluster.ClusterActorRefProvider, Akka.Cluster"

                    }

                    remote {
                        log-remote-lifecycle-events = DEBUG

                        helios.tcp {
                            transport-class = "Akka.Remote.Transport.Helios.HeliosTcpTransport, Akka.Remote"
                            applied-adapters = []
                            transport-protocol = tcp
                            #will be populated with a dynamic host-name at runtime if left uncommented
                            #public-hostname = "POPULATE STATIC IP HERE"
                            hostname = "127.0.0.1"
                            port = 8080
                        }
                    }            

                    cluster {
                        #will inject this node as a self-seed node at run-time
                        seed-nodes = ["akka.tcp://akkaconcert@127.0.0.1:8080"] #manually populate other seed nodes here, i.e. "akka.tcp://lighthouse@127.0.0.1:4053", "akka.tcp://lighthouse@127.0.0.1:4044"
                        roles = [cluster]
                        auto-down-unreachable-after = 10s
                    }
                }
        ]]>
</hocon>

谢谢！

4

1 回答 1

3

问题出在消息大小上，消息太大而被丢弃。

接收更大消息的配置：

akka {
helios.tcp {
    # Maximum frame size: 4 MB
    maximum-frame-size = 4000000b
}

}

于 2015-08-26T20:31:18.967 回答