Akka 持久化查询事件流和 CQRS

我正在尝试在我的 ES-CQRS 架构中实现读取端。假设我有一个这样执着的演员：

object UserWrite {

  sealed trait UserEvent
  sealed trait State
  case object Uninitialized extends State
  case class User(username: String, password: String) extends State
  case class AddUser(user: User)
  case class UserAdded(user: User) extends UserEvent
  case class UserEvents(userEvents: Source[(Long, UserEvent), NotUsed])
  case class UsersStream(fromSeqNo: Long)
  case object GetCurrentUser

  def props = Props(new UserWrite)
}

class UserWrite extends PersistentActor {

  import UserWrite._

  private var currentUser: State = Uninitialized

  override def persistenceId: String = "user-write"

  override def receiveRecover: Receive = {
    case UserAdded(user) => currentUser = user
  }

  override def receiveCommand: Receive = {
    case AddUser(user: User) => persist(UserAdded(user)) {
      case UserAdded(`user`) => currentUser = user
    }
    case UsersStream(fromSeqNo: Long) => publishUserEvents(fromSeqNo)
    case GetCurrentUser => sender() ! currentUser
  }

  def publishUserEvents(fromSeqNo: Long) = {
    val readJournal = PersistenceQuery(context.system).readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)
    val userEvents = readJournal
      .eventsByPersistenceId("user-write", fromSeqNo, Long.MaxValue)
      .map { case EventEnvelope(_, _, seqNo, event: UserEvent) => seqNo -> event }
    sender() ! UserEvents(userEvents)
  }
}

据我了解，每次事件持续存在时，我们都可以通过以下方式发布它Akka Persistence Query。现在，我不确定订阅这些事件的正确方法是什么，以便我可以将其保留在我的读取端数据库中？其中一个想法是首先发送一个UsersStream我的读方演员发来的消息UserWrite演员和“沉没”事件中读取演员。

EDIT

根据@cmbaxter的建议，我以这种方式实现了读取端：

object UserRead {

  case object GetUsers
  case class GetUserByUsername(username: String)
  case class LastProcessedEventOffset(seqNo: Long)
  case object StreamCompleted

  def props = Props(new UserRead)
}

class UserRead extends PersistentActor {
  import UserRead._

  var inMemoryUsers = Set.empty[User]
  var offset        = 0L

  override val persistenceId: String = "user-read"

  override def receiveRecover: Receive = {
    // Recovery from snapshot will always give us last sequence number
    case SnapshotOffer(_, LastProcessedEventOffset(seqNo)) => offset = seqNo
    case RecoveryCompleted                                 => recoveryCompleted()
  }

  // After recovery is being completed, events will be projected to UserRead actor
  def recoveryCompleted(): Unit = {
    implicit val materializer = ActorMaterializer()
    PersistenceQuery(context.system)
      .readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)
      .eventsByPersistenceId("user-write", offset + 1, Long.MaxValue)
      .map {
        case EventEnvelope(_, _, seqNo, event: UserEvent) => seqNo -> event
      }
      .runWith(Sink.actorRef(self, StreamCompleted))
  }

  override def receiveCommand: Receive = {
    case GetUsers                    => sender() ! inMemoryUsers
    case GetUserByUsername(username) => sender() ! inMemoryUsers.find(_.username == username)
    // Match projected event and update offset
    case (seqNo: Long, UserAdded(user)) =>
      saveSnapshot(LastProcessedEventOffset(seqNo))
      inMemoryUsers += user
  }
}

存在一些问题，例如： 事件流似乎很慢。 IE。UserRead参与者可以在保存新添加的用户之前回答一组用户。

EDIT 2

我增加了 cassandra 查询日志的刷新间隔，这更多地解决了事件流缓慢的问题。看来 Cassandra 事件日志默认每 3 秒轮询一次。在我的application.conf我补充道：

cassandra-query-journal {
  refresh-interval = 20ms
}

EDIT 3

实际上，不要减少刷新间隔。这会增加内存使用量，但这并不危险，也没有什么意义。 CQRS 的一般概念是写入和读取端是异步的。因此，写入数据后永远无法立即读取。处理用户界面？我只是在读取方确认数据后打开流并通过服务器发送的事件推送数据。

有一些方法可以做到这一点。例如，在我的应用程序中，我的查询端有一个参与者，它有一个持续查找更改的 PersistenceQuery，但您也可以有一个具有相同查询的线程。问题是保持流打开，以便能够在持久事件发生时立即读取它

val readJournal =
PersistenceQuery(system).readJournalFor[CassandraReadJournal](
  CassandraReadJournal.Identifier)

// issue query to journal
val source: Source[EventEnvelope, NotUsed] =
  readJournal.eventsByPersistenceId(s"MyActorId", 0, Long.MaxValue)

// materialize stream, consuming events
implicit val mat = ActorMaterializer()
source.map(_.event).runForeach{
  case userEvent: UserEvent => {
    doSomething(userEvent)
  }
}

相反，您可以使用一个计时器来引发 PersistenceQuery 并存储新事件，但我认为打开流是最好的方法