Akka-persistence：能否将持久化事件在Actor间路由？

针对你遇到的问题——已持久化的ChildActor的Created事件需要被ParentActor处理以维护子Actor集合，以下是几种可行的实现方案：

方案一：离线数据迁移（推荐用于历史数据固化场景） #

编写一次性迁移工具，读取所有ChildActor的Created事件，转换为ParentActor的ChildCreated事件并写入其事件日志。这样ParentActor启动时可直接通过自身事件日志恢复完整状态，无需依赖ChildActor的事件。

实现步骤： #

1. 使用Akka Persistence Query API查询所有ChildActor的持久化ID（假设ID命名有规律，如前缀为child-）。
2. 遍历每个ChildActor的事件流，筛选出Created事件并转换为ChildCreated事件。
3. 将转换后的事件写入ParentActor的事件日志（可通过调用ParentActor的持久化方法，或直接操作Journal API）。

示例代码（Scala）： #

import akka.persistence.query.{PersistenceQuery, EventEnvelope}
import akka.persistence.cassandra.query.scaladsl.CassandraReadJournal

val system = akka.actor.ActorSystem("MigrationSystem")
val query = PersistenceQuery(system).readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)

// 获取所有ChildActor的持久化ID
val childPersistenceIds = query.currentPersistenceIds().filter(_.startsWith("child-"))

childPersistenceIds.runForeach { persistenceId =>
  // 查询该ChildActor的所有事件
  val events = query.eventsByPersistenceId(persistenceId, 0, Long.MaxValue)
  events.runForeach { envelope: EventEnvelope =>
    envelope.event match {
      case Created(childId, name) =>
        // 向ParentActor发送持久化命令
        val parentActor = system.actorSelection("/user/ParentActor")
        parentActor ! PersistChildCreated(childId, name)
    }
  }
}

// ParentActor中处理持久化命令
class ParentActor extends PersistentActor {
  override def persistenceId: String = "ParentActor"
  
  private var children: Set[String] = Set.empty
  
  def receiveCommand: Receive = {
    case PersistChildCreated(childId, name) =>
      persist(ChildCreated(childId, name)) { evt =>
        children += evt.childId
      }
    // 其他命令处理...
  }
  
  def receiveRecover: Receive = {
    case evt: ChildCreated => children += evt.childId
  }
}

优缺点： #

• 优点：不影响系统运行时性能，迁移后ParentActor完全独立于ChildActor的事件流。
• 缺点：需要暂停写入或处理并发写入的一致性问题，适合历史数据不再变更的场景。

方案二：ParentActor启动时主动查询并重放ChildActor事件 #

ParentActor启动时，通过Persistence Query API读取所有ChildActor的Created事件，转换后应用到自身状态。需注意幂等性，避免重复处理同一事件。

实现要点： #

1. ParentActor初始化时，查询所有ChildActor的事件流。
2. 对每个Created事件，转换为ChildCreated并持久化（或直接更新状态，建议持久化以保证重启后状态一致）。
3. 在ParentActor状态中记录已处理的childId，避免重复处理。

示例代码片段： #

class ParentActor extends PersistentActor {
  private var children: Set[String] = Set.empty
  
  override def preStart(): Unit = {
    super.preStart()
    val query = PersistenceQuery(context.system).readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)
    val childEvents = query.eventsByPersistenceId("child-1", 0, Long.MaxValue) // 可扩展为遍历所有child ID
    childEvents.runForeach { envelope =>
      envelope.event match {
        case Created(childId, name) if !children.contains(childId) =>
          persist(ChildCreated(childId, name)) { _ =>
            children += childId
          }
      }
    }
  }
  
  // 其他方法...
}

优缺点： #

• 优点：无需离线迁移，可实时同步ChildActor的历史事件。
• 缺点：每次启动都要查询事件流，数据量大时会影响启动速度；需处理并发场景下的事件重复问题。

方案三：ChildActor事件恢复时转发消息给ParentActor #

修改ChildActor的事件恢复逻辑，在重放Created事件时，将转换后的ChildCreated消息发送给ParentActor。需保证ParentActor已启动，并处理消息的幂等性。

实现要点： #

1. 在ChildActor的receiveRecover方法中，重放Created事件后，向ParentActor发送ChildCreated消息。
2. ParentActor收到消息时，先检查子Actor是否已存在，再更新状态（或持久化事件）。

示例代码： #

class ChildActor(childId: String) extends PersistentActor {
  override def persistenceId: String = s"child-$childId"
  
  override def receiveRecover: Receive = {
    case evt: Created =>
      updateState(evt)
      // 转发给ParentActor
      context.system.actorSelection("/user/ParentActor") ! ChildCreated(childId, evt.name)
  }
  
  // 其他方法...
}

class ParentActor extends PersistentActor {
  private var children: Set[String] = Set.empty
  
  def receiveCommand: Receive = {
    case ChildCreated(childId, name) if !children.contains(childId) =>
      persist(ChildCreated(childId, name)) { _ =>
        children += childId
      }
    // 其他命令处理...
  }
  
  // 其他方法...
}

优缺点： #

• 优点：无需额外迁移工具，利用现有Actor的恢复流程同步事件。
• 缺点：依赖ChildActor的重启触发事件转发，未重启的ChildActor无法同步历史事件；需处理ParentActor未启动时的消息丢失问题。

关键结论 #

Akka Persistence本身没有内置的跨Actor事件路由机制，但可以通过上述方案手动实现事件的跨Actor迁移/同步。选择方案时需考虑：

• 历史数据量大小
• 系统是否需要持续同步新事件
• 对运行时性能的影响

内容的提问来源于stack exchange，提问作者gervais.b