处理来自在 AKKA.NET 参与者的任务中运行的服务的事件

Question

我正在为我的 WPF 应用程序使用 Prism 框架。我有一个生产者服务，它在任务中运行，并在找到文件时引发 CompositePresentationEvent。我有一个订阅该事件的 Akka 演员。 actor的处理程序看起来很简单，看起来像这样：

private void HandleFileReceive(FileEventArgs args)
{
    Self.Tell(new FileReceived(args.File));
}

当事件到达上面的处理程序时，我收到带有以下消息的 System.NotSupportedException：“没有活动的 ActorContext，这很可能是由于在此 Actor 中使用了异步操作”。

我认为这是因为服务在与参与者的处理程序不同的线程中运行。有没有办法在 Akka.NET 中处理这种类型的事情？

我并不完全反对编写新的 Actor 来完成我的情况所需的服务工作。问题是，根据文件中的某些设置，服务会有所不同。目前，我正在使用 MEF 处理这个问题，并从 IoC 容器中获取给定接口的正确实现者。我想继续将生产者的具体实现从核心代码（演员所在的地方）中抽象出来。

关于通过这个（感知的）线程问题和/或动态生成实现给定接口的 ProducerActor 的任何建议？

谢谢

-g

Answer 1

我有一个类似的问题，我想运行一个进程来运行来自参与者的命令行应用程序。问题是我想得到输出，这是通过处理process.OutputDataReceived 来完成的。

我最终做的是使用自定义堆栈来放置来自处理程序 process.OutputDataReceived += (sender, e) => Output.Push(e.Data); 的消息

自定义堆栈看起来像

class OutputStack<T> : Stack<T>
{
   public event EventHandler OnAdd;
   public void Push(T item)
   {
       base.Push(item);
       if (null != OnAdd) OnAdd(this, null);
   }
} 

然后在构造函数中我只处理自定义堆栈的OnAdd

OutputStack<string> Output = new OutputStack<string>();
Output.OnAdd += (sender, e) =>
{
    if (Output.Count > 0)
    {
        var message = Output.Pop();
        actor.Tell(new LogActor.LogMessage(message));
    }
};

这有点笨拙，但它可以工作，我可以在消息发生后立即发送（并得到处理）。希望我能在未来修复它...

Answer 2

我最终创建了一个参与者来处理响应检索并更改生产者接口，因此具体的生产者只需要实现一个简单的方法。旧的生产者负责在检查响应之前等待配置驱动的时间，但新代码利用 AKKA 的调度程序以相同的配置驱动间隔告诉检索器何时检查响应。我在下面提供了一个示例，但我的代码在错误处理等方面有点复杂。

在下面的代码中，您可以看到 IProducer 的接口和具体实现：LocalProducer 和 EmailProducer。在实际代码中，这些属性告诉容器一些其他信息，这些信息用于从容器中获取正确的实现。 ConsumerActor 是这个场景的父actor，负责处理Consume 消息。它在其构造函数中创建一个 ResponseRetrieverActor 并为 ResponseRetrieverActor 安排一个重复的通知，以便检索器在给定的时间间隔检查响应。

在 ResponseRetrieverActor 的 RetrieveResponses 处理程序中是 IoC 魔法发生的地方。我从工厂方法获取传输类型（此处未显示——它实际上是在各种配置文件中设置的）。传输类型用于使用具体生产者上的上述属性从 IoC 容器中检索正确的生产者。最后，生产者用于获取响应并将列表中的每个文件的信息告知父消费者，即消费者。

public interface IProducer
{
   List<string> GetResponses();
}

[Export(typeof(IProducer))] // Other attributes needed for MEF Export to differentiate the multiple IProducer implementations
public LocalProducer : IProducer
{
   public List<string> GetResponse()
   {
       // get files from a directory
   }
}

[Export(typeof(IProducer))]
public EmailProducer : IProducer
{
   public List<string> GetResponse()
   {
       // get files from email account
   }
}

public class ConsumerActor
{
    public class Consume
    {
       public Consume(string file) { this.File = file; }

       public string File { get; set; }
    }

    public ConsumerActor() 
    {
       _retriever = Context.ActorOf(Props.Create<ResponseRetrieverActor>(), "retriever");

       var interval = 10000;
       Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(0, interval, _retriever, new ResponseRetrieverActor.RetrieveResponses(), Self);
       Start(); 
    }

    private void Start()
    {
        Receive<Consume>(msg => 
        {
           // do something with the msg.File 
        });
    }

    private IActorRef _retriever;
}
public class ResponseRetrieverActor
{
  public class RetrieveResponses { }

  public ConsumerActor() { Start(); }

  private void Start()
  {
    Receive<RetrieveResponses>(msg => HandleRetrieveResponses());  
  }

  private void HandleRetrieveResponses()
  {
     var transportType = TransportFactory.GetTransportType(); // Gets transport protocol for the producer we need to use (Email, File, ect.)
     var producer = ServiceLocator.Current.GetInstance<IProducer>(transportType); // Gets a producer from the IoC container for the specified transportType

     var responses = producer.GetResponses();

     foreach(var response in responseFiles)
     {
         Context.Parent.Tell(new ConsumerActor.Consume(response));
     }
  }
}