akka-cluster-sharding：receive（）vs extarctShardId()和extractentityId（）

Question

我目前正在尝试修改使用Akka Actors实现的simple dist信仰框架的源代码。原始源代码在这里：http://alexminnaar.com/implementing-the-distbelieve-deep-neural-network-training-framework-with-akka.html。最初的实现只是基于Akka Actor，但我想将其扩展到分布式模式。我认为Akka-Cluster-Sharding是这个任务的正确选择。但是我想知道在哪里可以正确处理传入的消息，在receive（）方法中，或者在actor类中的extractShardId()&extractEntityId()中（例如对于ParameterShard actor，您可以在上面给定的链接中看到完整的源代码）。Akka的官方文档说:*extractEntityId和extractShardId是两个应用程序特定的函数，用于从传入消息中提取实体标识符和碎片标识符。

object ParameterShard {
  case class ParameterRequest(dataShardId: Int, layerId: Int)
  case class LatestParameters(weights: DenseMatrix[Double])
}

class ParamServer(shardId: Int,
                  numberOfShards: Int,
                  learningRate: Double,
                  initialWeight: LayerWeight) extends Actor with ActorLogging {


  val shardName: String = "ParamServer"

  val extractEntityId: ShardRegion.ExtractEntityId = {
      //case ps: ParameterRequest => (ps.dataShardId.toString, ps)

  }

  val extractShardId: ShardRegion.ExtractShardId = {
      //case ps: ParameterRequest => (ps.dataShardId % numberOfShards).toString
  }
  //weights initialize randomly
  var latestParameter: LayerWeight = initialWeight

  def receive = {

    //A layer corresponding to this shardId in some model replica has requested the latest version of the parameters.
    case ParameterRequest(shardId, layerId) => {
      log.info(s"layer ${layerId} weights read by model replica ${shardId}")
      context.sender() ! LatestParameters(latestParameter)
    }

    /*
    A layer corresponding to this shardId in some model replica has computed a gradient, so we must update our
    parameters according to this gradient.
    */
    case Gradient(g, replicaId, layerId) => {
      log.info(s"layer ${layerId} weights updated by model replica ${replicaId}")
      latestParameter = latestParameter + g.t * learningRate
    }

  }

}

Answer 1

我不知道任何关于distbelieve的东西，所以我不能识别哪个演员是由什么来识别的。所以，很抱歉我不能提供一个示例实现。但是让我们来做一个例子：您想要从不同的域获取/example url。所以你的演员看起来

class FetchActor(domain: String) extends Actor {
  def receive = {
    case Fetch => // do the job
  }
}

object FetchActor {
  case object Fetch
}

为了对该actor使用集群分片，请如下所示进行重构

class FetchActor extends Actor {
  def receive = {
    case Fetch(domain) => // do the job
  }
}

object FetchActor {
  case class Fetch(domain: String) // move parameter to message
  val entityIdExtractor = {
    case msg @ Fetch(domain) => (domain, msg)   // Which message FetchActor process
  }
  val shardIdExtractor = {
    case Fetch(domain) => domain   // What is shard id
  }
}

在该示例中，我决定使用整个域作为shardIdExtractor，但您的应用程序可能需要顶级域。

这就够了。只需启动FetchActor和集群分片，akka就会自动处理它的分发。