Akka持久性查询事件流和CQRS

问题描述

我正在尝试在我的ES-CQRS体系结构中实现读取端。假设我有一个这样的持久性参与者：

I'm trying to implement read side in my ES-CQRS architecture. Let's say I have a persistent actor like this:

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 发布它。现在，我不确定订阅这些事件的正确方法是什么，以便可以将其持久保存在我的读取数据库中？一种想法是首先从我的阅读侧actor向 UserWrite actor和接收器事件发送 UsersStream 消息

As far as I understand, each time when event gets persisted, we can publish it via Akka Persistence Query. Now, I'm not sure what would be a proper way to subscribe on these events so I can persist it in my read side database? One of the ideas is to initially send a UsersStream message from my read side actor to UserWrite actor and "sink" events in that read actor.

EDIT

在@cmbaxter的建议下，我这样实现了读取方：

Following suggestion of @cmbaxter, I implemented read side this way:

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
  }
}

有一些类似的问题：事件流似乎很慢。即 UserRead actor可以在保存新添加的用户之前用一组用户回答。

There are some issues like: Event stream seems to be slow. I.e. UserRead actor can answer with set of users before the newly added user is being saved.

编辑2

我增加了cassandra查询日志的刷新间隔，这更不会解决慢事件流带来的问题。默认情况下，似乎Cassandra事件日志是每3秒进行一次轮询。在我的 application.conf 中，我添加了：

I increased refresh interval of cassandra query journal which more less solved issue with slow event stream. It appears that Cassandra event journal is by default, being polled each 3 seconds. In my application.conf I added:

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

编辑3

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

Actually, do not decrease refresh interval. That will increase memory usage but that's not dangerous, neither a point. In general concept of CQRS is that write and read side are async. Therefore, after you write data will never be available immediately for reading. Dealing with UI? I just open the stream and push data via server sent events after the read side acknowledges them.

推荐答案

有一些方法可以执行这个。例如，在我的应用程序中，我在查询端有一个参与者，该参与者具有一个PersistenceQuery，它一直在寻找更改，但是您也可以使用相同的查询来创建一个线程。事情是保持流打开，以便能够在发生持久事件时立即读取

There are some ways to do this. For example, in my app i have an actor in my query side that have a PersistenceQuery that is consistently looking for changes, but you can have a thread with the same query too. The thing is to maintain the stream open to be able to read the persisted event as soon as it happens

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并存储新事件的计时器，但是我认为打开流是最好的方法

Instead of this, you can have a timer that raises a PersistenceQuery and stores new events, but i think that having a stream open is the best way

这篇关于Akka持久性查询事件流和CQRS的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！