multithreading - Akka参与者之间的工作负载平衡

Question

我有2个Akka actor用于爬网链接，即在X页面中找到所有链接，然后在X中链接的所有页面中找到所有链接，以此类推...

我希望他们以相同的速度或多或少地进步，但是他们中的一个经常会挨饿，而另一个人会消耗所有资源。

我尝试了以下方法(简化)。
单页爬网由以下参与者完成:

class Crawler extends Actor {
  def receive = {
    case Crawl(url, kind) =>
      // download url
      // extract links
      sender ! Parsed(url, links, kind)
  }
}

方法1:

class Coordinator extends Actor {
  val linksA = ...
  val linksB = ...
  def receive = {
    case Parsed(url, links, kind) =>
      val store = if (kind == kindA) linksA else linksB
      val newLinks = links -- store
      store ++= links
      newLinks.foreach { link =>
        val crawler = context.actorOf(Props[Crawler])
        crawler ! Crawl(link, kind)
      }
  }
}

方法2:

class Coordinator extends Actor {
  val linksA = ...
  val linksB = ...
  val rrProps = Props[Crawler].withRouter(RoundRobinRouter(nrOfInstances = 10)
  val crawlerA = context.actorOf(rrProps)
  val crawlerB = context.actorOf(rrProps)
  def receive = {
    case Parsed(url, links, kind) =>
      val store = if (kind == kindA) linksA else linksB
      val newLinks = links -- store
      store ++= links
      newLinks.foreach { link =>
        if (kind == kindA) crawlerA ! Crawl(link, kind)
        else crawlerB ! Crawl(link, kind)
      }
  }
}

第二种方法使情况稍好一些，但并不能完全解决问题。

是否有一种很好的方法可以使这两种类型的爬虫以相同的速度或多或少地前进？我是否应该在它们之间发送消息，从而彼此解除封锁？

Answer 1

我正在开发一个类似的程序，其中工作人员的资源成本不一致(在我的情况下，任务是执行数据库查询并将结果转储到另一个数据库中，但是就像爬网不同的网站将具有不同的成本一样，也将有所不同)查询的费用不同)。我采用的两种处理方法:

用RoundRobinRouter替换SmallestMailboxRouter

不要让Coordinator一次发送其所有消息-而是分批发送，在您的情况下，您有十个工作人员，因此发送40条消息应使它们最初保持繁忙。每当工作人员完成任务时，它都会向Coordinator发送一条消息，此时Coordinator会发出另一条消息，该消息可能会发送给刚刚完成任务的工作人员。 (您也可以分批执行此操作，即在收到n“任务完成”消息后，Coordinator会发送另一条n消息，但不要使n太高，否则某些任务非常短的工作人员可能会闲置。)

第三种选择是作弊并在所有参与者之间共享一个 ConcurrentLinkedQueue:在填充队列之后， Coordinator向工作人员发送“开始”消息，然后工作人员轮询队列，直到队列为空。