Akka Classic Cluster Metrics 扩展实战指南：集群指标采集、自适应负载均衡与 Sigar 配置

后端并发编程异步编程【免费下载链接】akka-coreA platform to build and run apps that are elastic, agile, and resilient. SDK, libraries, and hosted environments.项目地址https://gitcode.com/gh_mirrors/ak/akka-core点击查看免费下载导读本文基于 Akka 官方文档 cluster-metrics.md 并结合akka-cluster-metrics模块源码系统讲解 Classic经典 Actor API集群指标扩展Cluster Metrics Extension的工作原理与实战用法。你将掌握如何启用指标扩展并订阅ClusterMetricsChanged事件、内置SigarMetricsCollector与JmxMetricsCollector两种采集器的取舍、Hyperic Sigar 原生库的三种供给方式以及如何基于剩余容量权重构建AdaptiveLoadBalancingPool/Group自适应负载均衡路由器。文中所有配置参数均以仓库内 reference.conf 为准示例代码取自 akka-docs/src/test 下的可运行文档用例。模块引入与扩展启用添加依赖使用 Cluster Metrics 扩展需要在项目中引入akka-cluster-metrics依赖。以 Akka BOMakka-bom_$scala.binary.version$统一管理版本AkkaVersion 即为当前仓库对应的 Akka 版本// sbt libraryDependencies com.typesafe.akka %% akka-cluster-metrics % AkkaVersion!-- Maven -- dependency groupIdcom.typesafe.akka/groupId artifactIdakka-cluster-metrics_2.13/artifactId version${akka.version}/version /dependency// Gradle dependencies { implementation com.typesafe.akka:akka-cluster-metrics_2.13:$akkaVersion }注册扩展扩展是一个普通 Akka 扩展Extension需要在application.conf中显式注册。在 reference.conf 中扩展的入口类为akka.cluster.metrics.ClusterMetricsExtensionakka.extensions [ akka.cluster.metrics.ClusterMetricsExtension ]从源码看ClusterMetricsExtension继承自ExtensionIdProvider见 ClusterMetricsExtension.scala扩展创建时会启动一个名为cluster-metrics的监督者 ActorClusterMetricsSupervisor它负责按需创建/停止采集器子 Actor并可通过CollectionStartMessage/CollectionStopMessage控制消息在运行时启停采集。工作原理采集、平滑与散布职责划分根据 ClusterMetricsExtension.scala 的类注释扩展承担三项职责以固定频率采样本节点系统健康指标——采样委托给MetricsCollector对高方差数据做平滑处理——委托给EWMA指数加权移动平均算法把最新指标数据散布到节点环gossip并发布到本地事件总线——供自适应路由器与自定义监听者消费。集群成员状态为WeaklyUp的节点若启用该特性同样参与指标采集与散布参考 typed/cluster-membership.md#weaklyup-members。周期任务与 gossip核心执行者是ClusterMetricsCollectorActor见 ClusterMetricsCollector.scala。它在启动时调度两个周期任务MetricsTick采样按akka.cluster.metrics.collector.sample-interval默认 3s调用collector.sample()将本节点最新指标并入latestGossip并发布到本地事件总线GossipTick散布按akka.cluster.metrics.collector.gossip-interval默认 3s从节点环中随机挑选一个对等节点把MetricsGossipEnvelope发送过去收到对端 gossip 后通过latestGossip.merge(otherGossip)合并并回发replytrue的 gossip 包。因此每个节点掌握的是“本节点最新采样 采样间隔内收到的其他成员 gossip”ClusterMetricsChanged事件的有效载荷正是这组Set[NodeMetrics]。指标采集器Metrics Collector采集器接口与优先级指标采集被抽象为akka.cluster.metrics.MetricsCollectortrait只有一个方法sample(): NodeMetrics见 MetricsCollector.scala。扩展内置两个采集器实现实现类数据来源丰富度/精度前提条件akka.cluster.metrics.SigarMetricsCollectorHyperic Sigar 原生库 JMX更丰富、更精确需要 Sigar 类与原生库供给provisioningakka.cluster.metrics.JmxMetricsCollectorJMX MBeanMemoryMXBean、OperatingSystemMXBean相对有限无额外依赖作为兜底采集器实例的创建逻辑在MetricsCollector.applyMetricsCollector.scala中按以下优先级依次尝试配置的自定义采集器akka.cluster.metrics.collector.provider内置SigarMetricsCollector内置JmxMetricsCollector。配置项collector.fallback true默认时按“1 → 2 → 3”完整链路回退fallback false时只加载配置的自定义采集器失败即抛ConfigurationException。值得注意的是provider 且fallback true时会跳过第 1 步直接尝试 Sigar 再回退到 JMX。注意未供给 Sigar 时某些消息路由与 let-it-crash 类功能可能无法正常工作因为 CPU 相关指标combined/stolen只能由 Sigar 采集器提供。两类采集器的指标差异JmxMetricsCollectorMetricsCollector.scala采集systemLoadAverage过去 1 分钟系统负载JMX 不可用时为 -1 并被过滤、heapUsed、heapCommitted、heapMax、processors。注意系统负载与 CPU 利用率并非同一概念且 JMX 无法给出 CPU 时间片细分。SigarMetricsCollectorMetricsCollector.scala在 JMX 指标基础上叠加 Sigar 提供的cpuCombinedUser Sys Nice Wait 之和的百分比与cpuStolen虚拟化环境中被 hypervisor 抢占的 CPU 时间百分比。其构造器会在构造期主动调用metrics()验证 Sigar 可正常工作若类或原生库缺失则直接抛异常这正是回退机制被触发的时机。自定义采集器只需实现MetricsCollector并提供一个以ActorSystem为参数的公开构造器即可通过collector.provider配置接入。指标事件与订阅扩展按collector.sample-interval周期向节点系统事件总线发布akka.cluster.metrics.ClusterMetricsChanged快照事件每个采样周期发布一次见 ClusterMetricsCollector.scala。事件负载是Set[NodeMetrics]可通过getNodeMetrics()Java API遍历。订阅/退订由ClusterMetricsExtension的subscribe/unsubscribe方法完成其底层就是eventStream.subscribe(listener, classOf[ClusterMetricsEvent])见 ClusterMetricsExtension.scala。Scala 订阅示例仓库文档用例 MetricsListener.scala 展示了一个完整监听者在preStart订阅、postStop退订收到ClusterMetricsChanged后过滤出本节点指标并打印堆与负载信息class MetricsListener extends Actor with ActorLogging { val selfAddress Cluster(context.system).selfAddress val extension ClusterMetricsExtension(context.system) // Subscribe unto ClusterMetricsEvent events. override def preStart(): Unit extension.subscribe(self) // Unsubscribe from ClusterMetricsEvent events. override def postStop(): Unit extension.unsubscribe(self) def receive { case ClusterMetricsChanged(clusterMetrics) clusterMetrics.filter(_.address selfAddress).foreach { nodeMetrics logHeap(nodeMetrics) logCpu(nodeMetrics) } case state: CurrentClusterState // Ignore. } def logHeap(nodeMetrics: NodeMetrics): Unit nodeMetrics match { case HeapMemory(address, timestamp, used, committed, max) log.info(Used heap: {} MB, used.doubleValue / 1024 / 1024) case _ // No heap info. } def logCpu(nodeMetrics: NodeMetrics): Unit nodeMetrics match { case Cpu(address, timestamp, Some(systemLoadAverage), cpuCombined, cpuStolen, processors) log.info(Load: {} ({} processors), systemLoadAverage, processors) case _ // No cpu info. } }Java 订阅示例对应的 Java 版本见 MetricsListener.java使用ClusterMetricsExtension.get(system)获取扩展、receiveBuilder匹配事件并通过StandardMetrics.extractHeapMemory(nodeMetrics)/StandardMetrics.extractCpu(nodeMetrics)提取标准指标public class MetricsListener extends AbstractActor { LoggingAdapter log Logging.getLogger(getContext().getSystem(), this); Cluster cluster Cluster.get(getContext().getSystem()); ClusterMetricsExtension extension ClusterMetricsExtension.get(getContext().getSystem()); Override public void preStart() { extension.subscribe(getSelf()); } Override public void postStop() { extension.unsubscribe(getSelf()); } Override public Receive createReceive() { return receiveBuilder() .match(ClusterMetricsChanged.class, clusterMetrics - { for (NodeMetrics nodeMetrics : clusterMetrics.getNodeMetrics()) { if (nodeMetrics.address().equals(cluster.selfAddress())) { logHeap(nodeMetrics); logCpu(nodeMetrics); } } }) .match(CurrentClusterState.class, message - { /* Ignore. */ }) .build(); } }提示监听者 Actor 还应处理CurrentClusterState集群当前状态因为订阅事件总线后可能会先收到状态快照。Hyperic Sigar 供给ProvisioningSigar 使用操作系统原生库需要在运行时完成原生库的部署、解压与加载。官方文档与 reference.conf 顶部注释给出了三种供给方式以项目依赖方式引入 Kamon sigar-loaderio.kamon:sigar-loader扩展会按需借助 Kamon provisioner 解压并加载 sigar 库以 Java Agent 方式运行java -javaagent:/path/to/sigar-loader.jar在 JVM 启动阶段即完成解压与加载手动管理将sigar.jar放到classpath并将对应操作系统的原生库放到java.library.path项目依赖与库部署均由用户自己负责。引入 sigar-loader 的依赖声明如下// sbt libraryDependencies io.kamon %% sigar-loader % $sigar_loader.version$多实例部署的注意事项警告当使用 Kamon sigar-loader 且同一台主机上运行同一应用的多个实例时必须确保 sigar 库被解压到每个实例唯一的目录。可通过配置项akka.cluster.metrics.native-library-extract-folder控制解压目录。reference.conf中该配置默认值为${user.dir}/native即程序工作目录下的native子目录。官方建议使用按应用实例作用域隔离的位置如各实例的工作目录避免多实例互相覆盖已加载的原生库。自适应负载均衡Adaptive Load Balancing集群指标的主要消费方是自适应负载均衡路由器。AdaptiveLoadBalancingPool池与AdaptiveLoadBalancingGroup组根据集群指标数据以“节点剩余容量推导出的概率”对消息做随机选择路由见 ClusterMetricsRouting.scala。核心路由逻辑AdaptiveLoadBalancingRoutingLogic内部维护一个WeightedRoutees它把各节点权重累加为桶数组用Arrays.binarySearchThreadLocalRandom做加权随机挑选ClusterMetricsRouting.scala。权重由MetricsSelector根据最新ClusterMetricsChanged事件实时更新。内置 MetricsSelector 与权重公式reference.conf将cluster-metrics-adaptive-pool/cluster-metrics-adaptive-group映射到两个路由器类见 reference.confmetrics-selector配置值在MetricsSelector.fromConfig中解析ClusterMetricsRouting.scala。配置值选择器类依据指标剩余容量公式heapHeapMetricsSelector已用/最大 JVM 堆(max - used) / maxmax 缺失时退化为(committed - used) / committedloadSystemLoadAverageMetricsSelector过去 1 分钟系统负载均值对应 Linuxtop中的 load average1 - min(1.0, load / processors)负载接近 CPU 核数说明逼近瓶颈cpuCpuMetricsSelectorCPU 利用率百分比User Sys Nice Wait 之和叠加 stolen 权重1 - (combined stolen * (1.0 factor))其中factor 0.3对 stolen 时间加权mixMixMetricsSelectorheap、cpu、load 三者的组合各选择器剩余容量的平均值按地址聚合后求均值权重转换规则见CapacityMetricsSelector.weightsClusterMetricsRouting.scala容量最低的节点权重为 1最低可用容量按 1% 计同时避免除零其余节点按容量比例放大采集到的指标值会先经EWMA指数加权移动平均平滑平滑速率由collector.moving-average-half-life控制数值越小历史数据衰减越快、越偏向新数据。每个数据样本的相关性每经过一个半衰期减半4 个半衰期后降到约 6%。此外MetricsSelector允许完全自定义实现akka.cluster.metrics.MetricsSelector接口即可配置里填写 FQCN 时要求该类提供以com.typesafe.config.Config为参数的构造器。实战示例阶乘计算集群仓库文档用例用“计算阶乘”演示了该路由器的完整用法。后端工作者 FactorialBackend.scalaJava 版 FactorialBackend.java在 Future 中计算阶乘并回传结果class FactorialBackend extends Actor with ActorLogging { import context.dispatcher def receive { case (n: Int) Future(factorial(n)).map { result (n, result) }.pipeTo(sender()) } def factorial(n: Int): BigInt { tailrec def factorialAcc(acc: BigInt, n: Int): BigInt { if (n 1) acc else factorialAcc(acc * n, n - 1) } factorialAcc(BigInt(1), n) } }前端 FactorialFrontend.scalaJava 版 FactorialFrontend.java通过FromConfig.props()按配置创建路由器并把任务分发给后端class FactorialFrontend(upToN: Int, repeat: Boolean) extends Actor with ActorLogging { val backend context.actorOf(FromConfig.props(), name factorialBackendRouter) override def preStart(): Unit { sendJobs() if (repeat) context.setReceiveTimeout(10.seconds) } def receive { case (n: Int, factorial: BigInt) if (n upToN) { log.debug({}! {}, n, factorial) if (repeat) sendJobs() else context.stop(self) } case ReceiveTimeout log.info(Timeout) sendJobs() } def sendJobs(): Unit { log.info(Starting batch of factorials up to [{}], upToN) (1 to upToN).foreach { backend ! _ } } }路由器的配置注意只有router类型与metrics-selector是此扩展特有的其余字段与普通路由器一致akka.actor.deployment { /factorialFrontend/factorialBackendRouter { # Router type provided by metrics extension. router cluster-metrics-adaptive-group # Router parameter specific for metrics extension. # metrics-selector heap # metrics-selector load # metrics-selector cpu metrics-selector mix # routees.paths [/user/factorialBackend] cluster { enabled on use-roles [backend] allow-local-routees off } } }同一类型路由器也可在代码中定义Group查找式使用HeapMetricsSelector见 FactorialFrontend.scala#L68-L84import akka.cluster.routing.ClusterRouterGroup import akka.cluster.routing.ClusterRouterGroupSettings import akka.cluster.metrics.AdaptiveLoadBalancingGroup import akka.cluster.metrics.HeapMetricsSelector val backend context.actorOf( ClusterRouterGroup( AdaptiveLoadBalancingGroup(HeapMetricsSelector), ClusterRouterGroupSettings( totalInstances 100, routeesPaths List(/user/factorialBackend), allowLocalRoutees true, useRoles Set(backend))).props(), name factorialBackendRouter2)Pool部署式使用SystemLoadAverageMetricsSelector见 FactorialFrontend.scala#L89-L104import akka.cluster.routing.ClusterRouterPool import akka.cluster.routing.ClusterRouterPoolSettings import akka.cluster.metrics.AdaptiveLoadBalancingPool import akka.cluster.metrics.SystemLoadAverageMetricsSelector val backend context.actorOf( ClusterRouterPool( AdaptiveLoadBalancingPool(SystemLoadAverageMetricsSelector), ClusterRouterPoolSettings( totalInstances 100, maxInstancesPerNode 3, allowLocalRoutees false, useRoles Set(backend))).props(Props[FactorialBackend]()), name factorialBackendRouter3)从源码看路由器会通过内部的AdaptiveLoadBalancingMetricsListenerActor 订阅ClusterMetricsChanged事件并刷新权重ClusterMetricsRouting.scala因此无需额外配置即可自动感知指标变化。自定义 Metrics Collector如果你需要采集内置采集器未提供的指标如磁盘 IO、网络吞吐等可以实现自己的MetricsCollector继承akka.cluster.metrics.MetricsCollector实现sample(): NodeMetrics提供以akka.actor.ActorSystem为参数的公开构造器从源码dynamicAccess.createInstanceForMetricsCollector)可确认此要求见 MetricsCollector.scala将 FQCN 配置到akka.cluster.metrics.collector.provider保持collector.fallback true或false以决定失败时是否回退到内置采集器。官方建议以SigarMetricsCollector与JmxMetricsCollector两个内置实现为参考模板MetricsCollector.scala。完整配置参考以下为akka.cluster.metrics的全部配置项默认值取自仓库 reference.confakka.cluster.metrics { # 调度器配置键全路径。 dispatcher akka.actor.default-dispatcher # 周期任务开始前的初始延迟。 periodic-tasks-initial-delay 1s # Sigar 原生库解压目录应使用按应用实例隔离的路径如程序工作目录。 native-library-extract-folder ${user.dir}/native # 指标监督者 Actor。 supervisor { # Actor 名称命名空间示例/system/cluster-metrics name cluster-metrics # 监督策略。 strategy { # 提供 akka.actor.SupervisorStrategy 的类 FQCN # 构造器签名需为 init(com.typesafe.config.Config)。 # 默认策略是 OneForOneStrategy 的可配置扩展。 provider akka.cluster.metrics.ClusterMetricsStrategy # 默认策略提供者的配置覆盖 provider 时需同步替换。 configuration { # 记录重启尝试。 loggingEnabled true # 子 Actor 失败重启的时间窗口。 withinTimeRange 3s # 窗口内最大重启次数超出后停止子 Actor。 maxNrOfRetries 3 } } } # 指标采集器 Actor。 collector { # 是否启用采集用于负载均衡节点。 # 也可在运行时通过向 /system/cluster-metrics 发送 # akka.cluster.metrics.{CollectionStartMessage,CollectionStopMessage} 控制。 enabled on # 采集器实现 FQCN需实现 akka.cluster.metrics.MetricsCollector # 并提供以 akka.actor.ActorSystem 为参数的公开构造器。 # 加载优先级1) 自定义采集器 2) SigarMetricsCollector 3) JmxMetricsCollector provider # 是否尝试全部 3 个采集器提供者为 false 时自定义提供者失败即报错。 fallback true # 节点采样频率间隔越短采集越频繁 # 同时控制指标发布到节点系统事件总线的频率。 sample-interval 3s # 节点向集群其他节点发布指标 gossip 的频率。 gossip-interval 3s # EWMA 指数加权移动平均的半衰期控制历史数据相对新数据的衰减速度 # 值越低越偏向新数据。每个样本相关性每过一个半衰期减半 # 4 个半衰期后降至约 6%。初始相关性为 1 - 0.5^(collect-interval / half-life)。 moving-average-half-life 12s } }序列化与路由类型映射扩展的远程消息与路由器通过 protobuf 序列化reference.conf在akka.actor命名空间下注册序列化器akka-cluster-metrics akka.cluster.metrics.protobuf.MessageSerializerid 10绑定ClusterMetricsMessage、AdaptiveLoadBalancingPool、MixMetricsSelector、各选择器单例路由类型映射cluster-metrics-adaptive-pool→AdaptiveLoadBalancingPoolcluster-metrics-adaptive-group→AdaptiveLoadBalancingGroup。对应的 protobuf 定义位于 ClusterMetricsMessages.proto消息序列化实现见 MessageSerializer.scala。典型应用场景小结自适应负载均衡将mix/cpu/load/heap选择器应用于AdaptiveLoadBalancingPool/Group让消息自动偏向剩余容量最大的节点是集群指标最核心的用途基于指标的节点生命周期管理订阅ClusterMetricsChanged实现高级策略例如官方文档提到的 Node Let-it-crash——当 CPU stolen 时间过高时主动重启或下线节点此类场景依赖 Sigar 提供的 stolen CPU 指标监控与告警监听事件总线输出各节点堆内存、系统负载、CPU 利用率等指标可结合日志或外部监控系统展示集群健康度。需要注意的是本文介绍的是Classic Actor API的集群指标扩展指标事件ClusterMetricsChanged通过本地系统事件总线发布监听者 Actor 需自行在preStart/postStop中完成订阅与退订并兼容处理CurrentClusterState等集群状态消息。赞分享后端并发编程异步编程【免费下载链接】akka-coreA platform to build and run apps that are elastic, agile, and resilient. SDK, libraries, and hosted environments.项目地址https://gitcode.com/gh_mirrors/ak/akka-core点击查看免费下载相关推荐Akka Classic Cluster 实战指南集群扩展、成员事件订阅、种子节点加入与运维配置Akka Classic Cluster 实战指南集群扩展、成员事件订阅、种子节点加入与运维配置 Akka Classic Cluster 是基于经典 Act后端并发编程异步编程Conan服务器水平扩展负载均衡与集群配置Conan服务器水平扩展负载均衡与集群配置 为什么需要Conan服务器集群 随着团队规模扩大和项目复杂度提升单一Conan服务器面临三大挑战并发请求瓶颈开发工具包管理器构建工具CLIEMQX Core/Replicant 集群负载均衡修复与 cluster core rebalance 命令实战指南EMQX Core/Replicant 集群负载均衡修复与 cluster core rebalance 命令实战指南 本篇文章基于 EMQX 开源仓库的变更记后端物联网消息队列通信创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考