AKKA.Net：如何在演员重启时从主管向孩子发送消息

如何解决AKKA.Net：如何在演员重启时从主管向孩子发送消息

考虑这个代码：

    import datetime as dt
    import matplotlib.pyplot as plt
    import matplotlib.dates as mdates
    from matplotlib.collections import polyCollection

    data = [(dt.datetime(2018,7,17,15),dt.datetime(2018,30),'sleep','nap_1'),(dt.datetime(2018,45),'eat','snack_1'),1,0),'work','meeting_1'),'nap_2'),'snack_2'),'project_2')]

    cats = {"sleep" : 1,"eat" : 2,"work" : 3}
    colormapping = {"sleep" : "C0","eat" : "C1","work" : "C2"}

    verts = []
    colors = []
    for d in data:
        v =  [(mdates.date2num(d[0]),cats[d[2]]-.4),(mdates.date2num(d[0]),cats[d[2]]+.4),(mdates.date2num(d[1]),cats[d[2]]-.4)]
        verts.append(v)
        colors.append(colormapping[d[2]])

    bars = polyCollection(verts,facecolors=colors)

    fig,ax = plt.subplots()
    ax.add_collection(bars)
    ax.autoscale()
    loc = mdates.MinuteLocator(byminute=[0,15,30,45])
    ax.xaxis.set_major_locator(loc)
    ax.xaxis.set_major_formatter(mdates.AutoDateFormatter(loc))

    ax.set_yticks([1,2,3])
    ax.set_yticklabels(["sleep","eat","work"])

    annot = ax.annotate("",xy=(0,xytext=(20,20),textcoords="offset points",bBox=dict(Boxstyle="round",fc="w"),arrowprops=dict(arrowstyle="->"))
    annot.set_visible(False)


    def update_annot(ind):
        # how to index the polycollection? get_offsets causes an error
        pos = bars.get_offsets()[ind["ind"][0]] 
        # causes 'IndexError: index 5 is out of bounds for axis 0 with size 1'
        annot.xy = pos
        # If we can get an index,can we use it to access the original 'data' to get the 
        # string value at d[3] from data ('nap_1') and use it to annotate?
        text = ""

        annot.set_text(text)
        annot.get_bBox_patch().set_facecolor(cmap(norm(c[ind["ind"][0]])))
        annot.get_bBox_patch().set_alpha(0.4)


    def hover(event):
        vis = annot.get_visible()
        if event.inaxes == ax:
            cont,ind = bars.contains(event)
            if cont:
                update_annot(ind)
                annot.set_visible(True)
                fig.canvas.draw_idle()
            else:
                if vis:
                    annot.set_visible(False)
                    fig.canvas.draw_idle()

    fig.canvas.mpl_connect("motion_notify_event",hover)

    plt.show()

子actor执行一些有时会失败的冒险操作 该调用使用了一些参数，这些参数是在创建子级之后从父级/主管 Actor 传递的。

如何“监督”这个场景？我需要在重新启动后处理相同的 public class ParentActor : ReceiveActor { public ParentActor() { for( i = 1; i <= 5; i++) var childActor = Context.ActorOf<ChildActor>($"Child{i}"); childActor.Tell( new LoadData(i)); } } public class ChildActor : ReceiveActor { public ChildActor() { Receive<LoadData>(msg => { var data = SomeRiskyOverNetworkCall(msg.AccountId); }); } } public class LoadData { public int AccountId { get; } public LoadData(int accountId) { AccountId = accountId; } } （具有相同的参数）消息。

解决方法

您可以使用带有一些主管策略的预重启钩子。下面是一个简单的例子。 https://getakka.net/articles/actors/fault-tolerance.html#creating-a-supervisor-strategy

using System;
using System.Threading;
using Akka.Actor;

namespace AkkaConsoleSimple
{
    public class Start
    {
        public Start()
        {
            
        }
        
    }

    public class DoSomething
    {
        public DoSomething(int who)
        {
            Who = who;
        }
        public int Who { get; private set; }
    }

    public class FailedMessage
    {
        public FailedMessage(object who)
        {
            Who = who;
        }
        public object Who { get; private set; }
    }

    public class Child : ReceiveActor
    {
        public Child()
        {
            Receive<DoSomething>(msg =>
            {
                Console.WriteLine($"getting message no {msg.Who}");
                if (msg.Who == 10)
                {
                    throw new StackOverflowException();
                }
                
            });
        }

        protected override void PreRestart(Exception cause,object message)
        {
            Sender.Tell(new FailedMessage(message));
            Self.Tell(message);
            base.PreRestart(cause,message);
        }
    }
    public class Parent : ReceiveActor
    {
        public Parent()
        {
            Receive<Start>(greet =>
            {
                var child = Context.ActorOf<Child>();
                for (int i = 0; i < 11; i++)
                {
                    var msg = new DoSomething(i);
                    child.Tell(msg);
                }
               
            });

            Receive<FailedMessage>(failed => Console.WriteLine(failed.Who));
        }
        protected override SupervisorStrategy SupervisorStrategy()
        {
            return new OneForOneStrategy(
                maxNrOfRetries: 10,withinTimeRange: TimeSpan.FromMinutes(1),localOnlyDecider: ex =>
                {
                    switch (ex)
                    {
                        case StackOverflowException ae:
                            return Directive.Restart;
                        default:
                            return Directive.Escalate;
                    }
                });
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            var system = ActorSystem.Create("MySystem");
            var greeter = system.ActorOf<Parent>("parent");
            greeter.Tell(new Start());

            Console.Read();
        }
    }
}