Akka.NET Cluster 扩展（Cluster Extension）实战指南：订阅集群事件与获取集群状态

后端微服务【免费下载链接】akka.netCanonical actor model implementation for .NET with local distributed actors in C# and F#.项目地址https://gitcode.com/gh_mirrors/ak/akka.net点击查看免费下载Akka.Cluster本质上是一个ActorSystem扩展通过Cluster.Get()即可在任意 Actor 中直接访问集群的成员信息与 gossip 数据。本文以官方SimpleClusterListener示例为主线系统讲解如何获取Cluster扩展引用、如何订阅/退订成员事件与可达性事件、两大类 gossip 事件类型的区别以及通过快照或事件流两种方式获取集群当前状态的完整方案帮助你在集群路由Cluster Router之外直接基于集群事件构建自定义的分布式逻辑。Cluster 扩展的本质一个 ActorSystem 扩展在 Akka.NET 中Akka.Cluster并不是独立运行的组件而是一个标准的ActorSystem扩展Extension。这一点在 Cluster.cs 中体现得非常明确ClusterExtension继承自ExtensionIdProviderCluster其CreateExtension方法负责在给定的ExtendedActorSystem上构造Cluster实例。从 Cluster 构造函数 可以看到扩展被创建时会完成一系列初始化工作加载ClusterSettings读取akka.cluster配置节校验当前ActorSystem的 Provider 是否为IClusterActorRefProvider否则抛出ConfigurationException这就是为什么必须把akka.actor.provider配置为cluster解析本节点的SelfUniqueAddress含 UID 的唯一地址创建失败检测器DefaultFailureDetectorRegistryAddress、downing 提供者在/system/cluster路径下创建ClusterDaemon守护 Actor 树并挂载ClusterReadView作为本地状态只读视图。获取 Cluster 扩展引用获取Cluster扩展的方式非常直接只需要调用静态方法Cluster.Get(ActorSystem)例如在一个 Actor 内部using Akka.Actor; using Akka.Cluster; using Akka.Event; namespace Samples.Cluster.Simple { public class SimpleClusterListener : UntypedActor { // get direct reference to the Cluster extension protected Akka.Cluster.Cluster Cluster Akka.Cluster.Cluster.Get(Context.System); } }该方法在 Cluster.cs 中的实现是system.WithExtensionCluster, ClusterExtension()即同一个ActorSystem只会存在一个Cluster扩展实例后续调用直接复用不会重复创建。拿到Cluster引用后你可以直接访问以下常用成员成员说明源码位置SelfAddress本节点的Addresshost:portCluster.csSelfUniqueAddress含 UID 的唯一地址UID 用于区分同一地址上重启的节点Cluster.csSelfRoles本节点所属的所有角色akka.cluster.roles集合Cluster.csState当前集群状态的快照CurrentClusterStateCluster.csSettings集群配置设置对象Cluster.csIsUp本节点是否已达到Up/WeaklyUp状态Cluster.csIsTerminated集群扩展是否已关闭Cluster.cs其中State属性并不是实时轮询得到的而是通过内部ClusterReadView持续订阅集群事件流维护的本地只读视图见 ClusterReadView.cs。订阅与退订集群事件GossipSubscribe订阅集群事件订阅集群事件使用Cluster.Subscribe(IActorRef subscriber, params Type[] to)订阅后该 Actor 会持续收到对应类型的集群域事件。示例// subscribe to all future IMemberEvents and get current state as snapshot Cluster.Subscribe(Self, new[] { typeof(ClusterEvent.IMemberEvent) });从 Subscribe 源码 可以确认两点该重载默认以快照模式InitialStateAsSnapshot发送当前集群状态作为第一条消息详见下文“获取集群状态”一节订阅类型参数有严格校验传入空数组或传入未实现IClusterDomainEvent接口的类型会抛出ArgumentException。因此订阅的事件类型必须是ClusterEvent.IClusterDomainEvent或其子类/接口。订阅消息最终通过ClusterCore.Tell(new InternalClusterAction.Subscribe(...))发送给/system/cluster守护 Actor并由 ClusterDomainEventPublisher 实际完成事件流订阅与状态回放。Unsubscribe退订集群事件Cluster.Unsubscribe(Self); // unsub from ALL events Cluster.Unsubscribe(Self, typeof(IMemberEvent)); // unsub from just IMemberEvent这两个Unsubscribe调用几乎相同区别在于第一个调用将Self从所有集群事件中退订不再接收任何IClusterDomainEvent第二个调用只退订Self对IMemberEvent类型事件的订阅其他类型如可达性事件仍会继续投递。其实现见 Cluster.cs退订null类型表示退订全部。订阅模式的最佳实践集群事件订阅通常在PreStart中发起在PostStop中退订以保证 Actor 重启后能重新订阅、停止后不会遗留无用的订阅者死信。官方示例正是这种写法protected override void PreStart() { // subscribe to IMemberEvent and UnreachableMember events Cluster.Subscribe(Self, ClusterEvent.InitialStateAsEvents, new []{ typeof(ClusterEvent.IMemberEvent), typeof(ClusterEvent.UnreachableMember) }); } protected override void PostStop() { Cluster.Unsubscribe(Self); }集群 Gossip 事件类型Gossip 事件大体可以分为三类成员事件Member events节点加入、离开、被移除等成员变化可达性事件Reachability events节点之间的连通性变化度量事件Metrics events目前尚未实现。事件类型的继承关系在 ClusterEvent.cs 中定义得十分清晰所有集群事件都实现IClusterDomainEvent标记接口该接口本身继承自IDeadLetterSuppression见 ClusterEvent.cs。成员事件IMemberEvent成员事件描述节点加入/离开/被移除等状态变化被 集群路由Cluster Router 用来自动调整路由目标列表。IMemberEvent接口定义在 ClusterEvent.cs其实现类包括事件类触发时机对应成员状态MemberJoined节点刚加入集群JoiningMemberWeaklyUp无法达成收敛时加入中的节点被提升为 WeaklyUpWeaklyUpMemberUp节点被 Leader 提升为 UpUpMemberLeft节点开始离开集群LeavingMemberExited节点完成退出流程ExitingMemberDowned节点被标记为 DownDownMemberRemoved节点最终从集群中移除携带PreviousStatus表示移除前状态Removed这些事件类都继承自MemberStatusChange抽象基类构造时会校验成员状态与事件类型一致见 ClusterEvent.cs。MemberRemoved是唯一携带PreviousStatus属性的事件——当PreviousStatus为Down时表示节点因不可达被 down 后移除为Exiting时表示优雅退出后移除见 ClusterEvent.cs。事件不是凭空产生的而是由ClusterEvent.DiffMemberEvents对比前后两次MembershipState计算差异生成的见 ClusterEvent.cs再由ClusterDomainEventPublisher在每次 gossip 变化后统一发布。可达性事件IReachabilityEvent可达性事件描述节点之间的连通性。如果节点 A 无法再到达节点 B那么 B 就被视为对 A 不可达unreachable当不可达节点数量较多时Leader 会将该节点标记为 down 并将其移出集群。反之如果 B 能重新与 A 通信A 会通过 gossip 宣布 B 重新 可达reachable。IReachabilityEvent接口定义在 ClusterEvent.cs实现类有两个UnreachableMember成员被失败检测器判定为不可达见 ClusterEvent.csReachableMember之前不可达的成员恢复连通见 ClusterEvent.cs。可达性事件的差异计算同样由内部DiffUnreachable/DiffReachable方法完成见 ClusterEvent.cs。其他集群域事件除成员与可达性事件外Cluster扩展还发布以下事件ClusterEvent.IClusterDomainEvent所有集群消息的统称接口订阅它等于订阅全部集群事件ClusterEvent.LeaderChanged集群 Leader 发生变化时触发携带新的 Leader 地址可能为 null见 ClusterEvent.csClusterEvent.RoleLeaderChanged某个角色集合内的 Leader 发生变化时触发同时携带角色名与新 Leader 地址见 ClusterEvent.csClusterShuttingDown整个集群扩展关闭通知很少发生由ClusterDomainEventPublisher.PostStop在关闭时发布单例事件见 ClusterEvent.cs。订阅这些事件后你就可以基于集群状态做出判断例如“某个角色的成员是否已经有 2 个处于 Up 状态”“Leader 是否不可达”等。完整示例SimpleClusterListener下面把SimpleClusterListener完整展开。它订阅了ClusterEvent.IMemberEvent与ClusterEvent.UnreachableMember后者实现了ClusterEvent.IReachabilityEvent并在OnReceive中逐一处理public class SimpleClusterListener : UntypedActor { protected ILoggingAdapter Log Context.GetLogger(); protected Akka.Cluster.Cluster Cluster Akka.Cluster.Cluster.Get(Context.System); /// summary /// Need to subscribe to cluster changes /// /summary protected override void PreStart() { // subscribe to IMemberEvent and UnreachableMember events Cluster.Subscribe(Self, ClusterEvent.InitialStateAsEvents, new []{ typeof(ClusterEvent.IMemberEvent), typeof(ClusterEvent.UnreachableMember) }); } /// summary /// Re-subscribe on restart /// /summary protected override void PostStop() { Cluster.Unsubscribe(Self); } protected override void OnReceive(object message) { var up message as ClusterEvent.MemberUp; if (up ! null) { var mem up; Log.Info(Member is Up: {0}, mem.Member); } else if(message is ClusterEvent.UnreachableMember) { var unreachable (ClusterEvent.UnreachableMember) message; Log.Info(Member detected as unreachable: {0}, unreachable.Member); } else if (message is ClusterEvent.MemberRemoved) { var removed (ClusterEvent.MemberRemoved) message; Log.Info(Member is Removed: {0}, removed.Member); } else if (message is ClusterEvent.IMemberEvent) { //IGNORE } else { Unhandled(message); } } }Cluster会随着集群成员与可达性的变化周期性地广播这些消息并作为普通 Actor 消息投递到SimpleClusterListener的OnReceive中处理。仓库中的实际示例还额外处理了MemberJoined与CurrentClusterState两种消息见 SimpleClusterListener.cs完整可运行的示例项目位于 src/examples/Cluster/Samples.Cluster.Simple。示例项目的 Program.cs 展示了如何用ConfigurationFactory.ParseString覆盖端口、以akka.actor.provider cluster开启集群并通过seed-nodes配置种子节点akka { actor { provider cluster } remote { log-remote-lifecycle-events DEBUG dot-netty.tcp { hostname localhost port 0 } } cluster { seed-nodes [ akka.tcp://ClusterSystemlocalhost:2551, akka.tcp://ClusterSystemlocalhost:2552] } }什么时候应该直接使用 Gossip现实当中大约 99% 的场景你会使用集群路由Cluster Router而不是直接消费 gossip 事件。只有集群路由不合适的场景才需要直接订阅 gossip例如图中列举的几种情况需要感知**角色role**维度的成员变化如“某个角色的成员已就绪”需要感知Leader 变化如LeaderChanged、RoleLeaderChanged需要感知可达性变化如 Leader 不可达需要自定义的成员发现、状态聚合或故障处理逻辑。一句话总结路由解决“把消息发给谁”gossip 事件解决“集群现在长什么样、发生了什么变化”。获取集群状态Cluster的当前状态可以通过两种途径获取订阅时的初始状态回放新订阅者会收到当前状态按需查询显式请求当前状态快照。订阅时获取初始状态Cluster.Subscribe的第二个重载允许你指定初始状态模式见 Cluster.cs// subscribe to all future IMemberEvents and get current state as snapshot Cluster.Subscribe(Self, ClusterEvent.SubscriptionInitialStateMode.InitialStateAsSnapshot, new[] { typeof(ClusterEvent.IMemberEvent) }); // subscribe to all future IMemberEvents and get current state as event stream Cluster.Subscribe(Self, ClusterEvent.SubscriptionInitialStateMode.InitialStateAsEvents, new[] { typeof(ClusterEvent.IMemberEvent) });两种调用完成的目标相同——将Self订阅到未来的所有IMemberEvent事件——区别在于Self如何收到集群的当前状态InitialStateAsSnapshot快照模式订阅者会收到一条ClusterEvent.CurrentClusterState消息其中包含当前全部成员、各自角色与成员状态Joining/Up/Leaving等。CurrentClusterState还暴露Unreachable不可达成员集合、SeenBy、Leader、RoleLeader(role)等属性见 ClusterEvent.cs。这也是Subscribe不带模式参数时的默认行为见 Cluster.cs。InitialStateAsEvents事件流模式订阅者会收到一串MemberUp、MemberLeft、MemberRemoved等历史事件——相当于以事件回放的方式重建“如果你从一开始就订阅会看到什么”的当前状态即事件溯源event-sourced版本的CurrentClusterState内容。从源码看事件流模式由ClusterDomainEventPublisher.Subscribe调用PublishDiff(_emptyMembershipState, _membershipState, Pub)实现即从“空状态”对比“当前状态”推导出应回放的事件序列见 ClusterEvent.cs快照模式则直接调用SendCurrentClusterState(subscriber)。需要注意订阅之后后续的成员变化事件MemberUp等仍会照常推送无论你选择哪种初始状态模式这些增量事件都需要在消息处理中覆盖到。按需获取集群状态订阅时回放的初始状态只在订阅那一刻生效之后不会自动重复推送完整状态。若需要随时获取当前状态的快照可以使用SendCurrentClusterState方法见 Cluster.cs// get the current state of cluster Cluster.SendCurrentClusterState(Self);该消息会异步地将一条ClusterEvent.CurrentClusterState投递给目标 Actor。如果你希望周期性刷新可以结合Scheduler定时调用源码注释也明确建议了这种用法。此外你也可以随时直接读取Cluster.State属性同步、非阻塞地拿到本地最新快照两者数据同源。仓库中的 ClusterSpec.cs 对按需获取状态做了直接验证调用SendCurrentClusterState后测试 Actor 能立即收到CurrentClusterState消息ClusterSpec.cs 则验证了InitialStateAsEvents模式下订阅者会先收到MemberUp事件流。这些测试用例可以作为你验证自己实现行为的参照。集群事件与 gossip 配置参数集群事件由底层 gossip 协议驱动。与事件发布节奏直接相关的配置位于 Cluster.conf配置项默认值说明akka.cluster.gossip-interval1s节点每隔多久向外发送一次 gossip 信息akka.cluster.gossip-time-to-live2s传入的 gossip 消息若在该时长内未被处理则被丢弃akka.cluster.gossip-different-view-probability0.8gossip 优先发送给持有更新/更旧状态视图的随机节点的概率0.0~1.0akka.cluster.reduce-gossip-different-view-probability400集群节点数超过该值时降低上述概率akka.cluster.periodic-tasks-initial-delay1s周期任务启动前的初始延迟akka.cluster.leader-actions-interval1sLeader 执行维护任务的频率akka.cluster.unreachable-nodes-reaper-interval1s将不可达节点移出成员环的清扫频率了解这些参数有助于理解事件到达的时机事件并不会“瞬间”到达所有节点而是依赖 gossip 周期性地扩散因此基于 gossip 构建的应用逻辑应当容忍一定的传播延迟。小结本文围绕Cluster这个ActorSystem扩展覆盖了从获取引用、订阅/退订事件、理解事件类型体系到通过快照/事件流获取集群状态的完整链路。核心要点回顾Akka.Cluster.Cluster.Get(Context.System)是访问集群扩展的统一入口Subscribe/Unsubscribe是消费集群事件的标准方式订阅类型必须是IClusterDomainEvent及其子类事件分为成员事件IMemberEvent系与可达性事件IReachabilityEvent系两大类另有LeaderChanged、RoleLeaderChanged、ClusterShuttingDown等域事件获取集群状态有两种方式订阅时选择InitialStateAsSnapshot快照或InitialStateAsEvents事件回放运行中通过SendCurrentClusterState或State属性按需读取直接消费 gossip 仅适用于集群路由无法覆盖的场景绝大多数情况下集群路由参见 集群路由仍是首选。如需进一步理解 gossip 协议本身的传播机制与集群形成过程可继续阅读 集群概览相关可运行示例请参考 Samples.Cluster.Simple。赞分享后端微服务【免费下载链接】akka.netCanonical actor model implementation for .NET with local distributed actors in C# and F#.项目地址https://gitcode.com/gh_mirrors/ak/akka.net点击查看免费下载相关推荐Akka.NET Cluster Client 完整指南让集群外的 ActorSystem 安全访问集群内 ActorAkka.NET Cluster Client 完整指南让集群外的 ActorSystem 安全访问集群内 Actor 导读 ClusterClient 是后端微服务Zeek Cluster 集群框架完全指南从 cluster-layout 布局到节点池、发布订阅与遥测Zeek Cluster 集群框架完全指南从 cluster layout 布局到节点池、发布订阅与遥测 Zeek 的 Cluster 框架 scripts网络安全网络IDSFeathers 应用横向扩展与集群部署实战从 Node.js Cluster 到多实例事件同步Feathers 应用横向扩展与集群部署实战从 Node.js Cluster 到多实例事件同步 本文是一份面向 Feathers feathersjs/后端Web框架上一篇electron-vue 生产与开发调试完全指南主进程远程调试5858 端口与生产构建排错实战下一篇Prisma 数据建模完全指南基于 GraphQL SDL 设计数据模型Data Modelling创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考