Akka Typed PubSub在K8s分布式集群跨节点消息同步异常问题

问题根源 #

你当前使用的akka.actor.typed.pubsub.Topic是本地Actor，每个K8s实例（Akka节点）都会独立创建一个名为ClusterMessage的Topic Actor实例。发布的消息只会在当前节点的Topic内部流转，自然无法被其他节点的订阅者接收。

解决方案 #

要实现集群范围内的Pub/Sub，需要改用Akka提供的集群版Topic，它基于Cluster Sharding实现，能在整个集群中维护一个全局唯一的Topic逻辑实例，所有节点的发布和订阅请求都会路由到这个实例。

步骤1：确认依赖

确保项目中引入了集群相关依赖（以sbt为例）：

libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-cluster-typed" % "2.8.5",
  "com.typesafe.akka" %% "akka-cluster-sharding-typed" % "2.8.5"
)

步骤2：修改代码使用集群版Topic

替换本地Topic为集群版akka.cluster.typed.pubsub.Topic，并确保Akka Cluster配置正确：

修改ClusterSystem

import akka.cluster.typed.pubsub.Topic
import akka.actor.typed.scaladsl.Behaviors
import akka.actor.typed.{ActorRef, ActorSystem}
import akka.cluster.typed.Cluster
import akka.cluster.sharding.typed.scaladsl.ClusterSharding
import org.slf4j.LoggerFactory

object ClusterSystem extends App {
  private val logger = LoggerFactory.getLogger(getClass)

  ActorSystem(
    Behaviors.setup[Unit] { context =>
      implicit val system: ActorSystem[_] = context.system
      implicit val ex: ExecutionContext = system.executionContext
      implicit val clusterSharding: ClusterSharding = ClusterSharding(system)

      // 监听集群节点就绪事件
      Cluster(system).registerOnMemberUp {
        logger.info("Cluster member ready, initializing pub/sub services")
      }

      // 创建集群全局唯一的Topic
      val pubSubRef: ActorRef[Topic.Command[Message]] = Topic(system, "cluster-message")

      val publishService = new PublisherService(pubSubRef)
      val subscribeService = new SubscribeService(pubSubRef)

      Behaviors.empty
    },
    "ClusterSystem"
  )
}

修改PublisherService（替换导入包即可）

import akka.cluster.typed.pubsub.Topic
import scala.concurrent.Future
import akka.Done

class PublisherService(pubSubRef: ActorRef[Topic.Command[Message]]) {
  def publish(): Future[Done] = {
    pubSubRef ! Topic.Publish(Message("someRandomData"))
    Future.successful(Done)
  }
}

修改SubscribeService（替换导入包即可）

import akka.cluster.typed.pubsub.Topic
import akka.stream.typed.scaladsl.PubSub
import akka.stream.scaladsl.Source
import akka.NotUsed

class SubscribeService(pubSubRef: ActorRef[Topic.Command[Message]]) {
  def subscribe(): Source[Message, NotUsed] = 
    PubSub.source(pubSubRef, bufferSize = 10, overflowStrategy = akka.stream.OverflowStrategy.dropHead)
}

步骤3：配置Akka Cluster（K8s环境）

在application.conf中配置集群发现和远程通信，确保节点能互相加入：

akka {
  actor.provider = cluster
  
  cluster {
    # 使用Kubernetes API进行服务发现
    discovery.method = kubernetes-api
    # 若用StatefulSet，可配置固定seed节点
    # seed-nodes = ["akka://ClusterSystem@cluster-system-0.default.svc.cluster.local:2551"]
    jmx.multi-mbeans-in-same-jvm = on
  }
  
  remote.artery.canonical {
    hostname = ${HOSTNAME} # K8s Pod的主机名，自动注入
    port = 2551
  }
}

关键说明 #

• 集群版Topic通过Cluster Sharding保证全局唯一性和高可用性：如果承载Topic的节点故障，Sharding会自动在其他存活节点重建Topic实例。
• 所有节点的发布请求都会被路由到全局Topic实例，订阅者会从该实例接收所有节点的发布消息，实现跨节点消息同步。

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