我有以下任务，对此我有 Java/Executors 解决方案运行良好，但我想在 Akka 中实现相同的功能并寻找最佳实践建议。

问题:

从多个 URL 中并行获取/解析数据，阻塞直到所有数据被获取并返回聚合结果。应该重试错误(IOException 等)达到一定次数。

到目前为止，我的实现非常简单——创建知道应该获取哪些 URL 的 Fetcher actor，它创建一堆 Worker actor 并向它们发送 URL，每条消息一个。完成特定 URL Worker 后，将消息连同结果发送回 Fetcher。 Fetcher 保持结果状态，Workers 无状态。下面是简化的代码。

getter :

class Fetcher extends UntypedActor {
  private ActorRef worker;

  public void onReceive(Object message) throws Exception {
    if (message instanceof FetchMessage) {
      this.worker = context().actorOf(SpringExtension.SpringExtProvider.get(actorSystem).props("Worker")
              .withRouter(new RoundRobinPool(4)), "worker");
      for(URL u: urls) {
        this.worker.tell(new WorkUnit(u), getSelf());
      }
   }
   else if (message instanceof Result) {
     // accumulate results
   }
}

worker :

class Worker extends UntypedActor {

  public void onReceive(Object message) throws Exception {
    if (message instanceof WorkUnit) {
      // fetch URL, parse etc
      // send result back to sender
      getSender().tell(new Result(...), null);
    }
}

到目前为止一切顺利，在没有异常(exception)的情况下，一切都按预期进行。

但是，如果在 Worker 中获取 URL 时出现 IOException，那么 Akka 将重启 Worker actor，但 Worker 正在处理的消息会丢失。即使我使用不同的 SupervisorStrategy，结果也是一样的——一些消息实际上“丢失”了。当然，我可以使用 try/catch 将代码包装在 Worker.onReceive() 中，但我觉得这违背了 Akka 的理念。我想我可以使用持久消息传递，但我认为在这种情况下增加消息持久性的复杂性是不合理的。

我可能需要某种方式让 Fetcher 发现 Worker 未能获取某些 URL 并再次重新发送 WorkUnit 或检测到某些结果在太长时间内没有返回。处理这种情况的最佳方法是什么？

谢谢，

最佳答案

我们在我们的项目中遇到了类似的问题，我们找到了一个适合我们的解决方案 - 无论异常、工作人员故障、网络故障等如何执行任务。尽管我必须承认代码最终变得有点复杂。

所以我们的设置如下:

1. 有一个 WorkerControl actor 处理任务管理和与 worker 的通信
2. 有许多 Worker actors 位于不同的 VM 中(可能位于不同的物理机器上)
3. WorkerControl 接收一些要处理的数据并在 worker 之间调度任务

我们或多或少地尝试遵循描述的指南 here

但我们也提高了设计的容错能力。

在 WorkerControl 中，我们保留以下数据结构:

Map<ActorPath, ActorRef> registeredWorkers // registry of workers
Deque<TaskInfo> todoList                   // tasks that have not been yet processed
Map<ActorRef, TaskInfo> assignedTasks      // tasks assigned to the workers
Map<ActorPath, ActorRef> deadWorkers       // registry of dead workers

对于每个要执行的任务，我们保留一个数据结构

class TaskInfo {
    private final WorkerTask task;
    private int failureCount = 0;
    private int restartCount = 1;
    private Date latestResultDelivery;
}

我们处理以下可能的故障列表

Worker 通过抛出异常(即您的情况下的 IOException)而使任务失败

我们向工作控件发送一条 new Failure(caughtException) 消息。看到它后，工作控件会增加 failureCount 并将任务放在 todoList 队列的头部。当达到给定的失败次数时，任务将被视为永久失败并且永远不会重试。 (之后，可以以自定义方式记录、处置和处理永久失败的任务)。

Worker在给定的时间段内没有交付任何结果(例如，他陷入死循环，worker机器上的资源争用，worker神秘消失，任务处理时间过长)

我们为此做了两件事

1. 我们初始化 taskInfo 的 latestResultDelivery 字段，并将任务分配存储在 assignedTasks 映射中。
2. 我们会定期对工作人员控件运行“健康检查”，以确定工作人员是否在某项任务上工作的时间过长。

    for (ActorRef busyWorker : assignedTasks.keySet()) {
        Date now = new Date();
        if (now.getTime()
                - assignedTasks.get(busyWorker).getLatestResultDeliveryTime() >= 0) {
            logger.warn("{} has failed to deliver the data processing result in time", nameOf(busyWorker));
            logger.warn("{} will be marked as dead", nameOf(busyWorker));
            getSelf().tell(new Failure(new IllegalStateException("Worker did not deliver any result in time")),
                    busyWorker);
            registeredWorkers.remove(busyWorker.path());
            deadWorkers.put(busyWorker.path(), busyWorker);
        }
    }

网络断开，工作进程死亡

我们又做了两件事:

1. 在 worker 向 worker 控件注册后，我们开始观察 worker Actor

   registeredWorkers.put(worker.path(), worker);
   context().watch(worker);

2. 如果我们在工作控件中收到一条 Terminated 消息，我们会增加 restartCount 并将任务返回到 todoList。同样，重启次数过多的任务最终会永久失败，并且再也不会重试。这是在任务本身成为远程工作人员死亡的原因(例如，由于 OutOfMemoryError 而导致远程系统关闭)的情况下完成的。我们为失败和重启保留单独的计数器，以便能够更好地精确重试策略。

我们还做了一些尝试，使 worker 本身具有容错能力。例如。工作人员控制他的任务的执行时间，并监控他最近是否一直在做任何事情。

根据您需要处理的故障类型，您可以实现所列策略的一部分。

底线:正如其中一条评论中提到的那样:为了重新安排任务，您需要在 Fetcher 中保留一些数据结构来映射工作人员和分配的任务。

关于java - 在 Akka actor 中处理异常的最佳实践，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/23945464/