如何获取Akka Cluster Sharding 2.5.23的活跃Actor数量？

你不需要依赖付费监控工具，Akka本身提供了几种轻量方式来统计活跃Actor数量，下面是具体实现方案：

方案1：利用Cluster Sharding内置状态查询 #

Akka Cluster Sharding内置了分片状态查询API，可以直接获取各分片的活跃实体数，汇总后得到总数。这种方式无需额外编码，适合临时查询或简单脚本场景。

import akka.cluster.sharding.ClusterSharding
import akka.pattern.ask
import akka.util.Timeout

import scala.concurrent.duration._
import scala.concurrent.Await

// 在ActorSystem上下文执行
implicit val timeout: Timeout = 10.seconds
val sharding = ClusterSharding.get(system)

// 获取所有分片的状态并汇总活跃实体数
val totalActiveActors = Await.result(sharding.shardState(), timeout.duration)
  .stats.values.map(_.numberOfEntities).sum

println(s"当前活跃Actor总数: $totalActiveActors")

说明：shardState()返回的是集群所有分片的状态，每个分片仅由一个节点负责，因此在任意节点执行该查询都能得到全集群的统计结果。

方案2：自定义计数器实时追踪 #

如果需要持续、实时的统计，可以通过Actor生命周期钩子实现自定义计数器，结合集群单例保证统计节点的可靠性。

步骤1：实现统计Actor #

import akka.actor.{Actor, ActorLogging, Props}

object ActorCounter {
  case object GetTotalCount
  case class IncrementCount(node: String)
  case class DecrementCount(node: String)
  case class TotalCount(count: Int)

  def props(): Props = Props(new ActorCounter())
}

class ActorCounter extends Actor with ActorLogging {
  private var nodeCounts: Map[String, Int] = Map.empty

  override def receive: Receive = {
    case ActorCounter.IncrementCount(node) =>
      nodeCounts = nodeCounts.updated(node, nodeCounts.getOrElse(node, 0) + 1)
    case ActorCounter.DecrementCount(node) =>
      nodeCounts = nodeCounts.updated(node, nodeCounts.getOrElse(node, 0) - 1)
    case ActorCounter.GetTotalCount =>
      sender() ! ActorCounter.TotalCount(nodeCounts.values.sum)
  }
}

步骤2：在实体Actor中绑定计数器 #

import akka.actor.Actor
import akka.cluster.Cluster

class MyEntityActor(counterRef: ActorRef) extends Actor {
  private val nodeAddress = Cluster(context.system).selfAddress.toString

  override def preStart(): Unit = {
    super.preStart()
    counterRef ! ActorCounter.IncrementCount(nodeAddress)
  }

  override def postStop(): Unit = {
    super.postStop()
    counterRef ! ActorCounter.DecrementCount(nodeAddress)
  }

  override def receive: Receive = {
    // 处理实体Actor的业务逻辑
    case msg => // 业务处理逻辑
  }
}

步骤3：启动Cluster Sharding并关联计数器 #

import akka.cluster.sharding.ClusterSharding
import akka.cluster.singleton.ClusterSingletonManager

// 以集群单例模式启动计数器Actor，避免单点故障
system.actorOf(
  ClusterSingletonManager.props(
    singletonProps = ActorCounter.props(),
    terminationMessage = PoisonPill,
    settings = ClusterSingletonSettings(system)
  ),
  name = "actorCounterSingleton"
)
val counterRef = ClusterSingletonProxy(system).proxyFor("/user/actorCounterSingleton")

// 启动Cluster Sharding时传递计数器引用
ClusterSharding(system).start(
  typeName = "MyEntity",
  entityProps = Props(new MyEntityActor(counterRef)),
  settings = ClusterShardingSettings(system),
  extractEntityId = msg => // 自定义实体ID提取逻辑,
  extractShardId = msg => // 自定义分片ID提取逻辑
)

步骤4：查询总数 #

import akka.pattern.ask
import akka.util.Timeout

implicit val timeout: Timeout = 5.seconds
val totalCountFuture = (counterRef ? ActorCounter.GetTotalCount).mapTo[ActorCounter.TotalCount]
totalCountFuture.foreach { count =>
  println(s"当前活跃Actor总数: ${count.count}")
}

方案3：通过JMX查看指标 #

Akka默认支持JMX暴露Cluster Sharding的指标，无需编码即可查看活跃实体数：

1. 在application.conf中启用JMX：

akka.jmx.enabled = on
akka.cluster.sharding.jmx.metrics.enabled = on

2. 使用JConsole或VisualVM连接到JVM，找到MBean akka:type=ClusterSharding,name=*，查看NumberOfEntities属性，汇总所有分片的数值即可得到总数。

内容的提问来源于stack exchange，提问作者Arun