akka - Akka 删除持久消息

Question

我正在编写一个应用程序来处理设备引发的事件（每小时 100 万）。一些事件将被聚合（并且具有很长的时间跨度（例如 48 小时）），其中包含开始事件、状态（x 次）事件和结束事件。其他是可以立即处理的单个事件。为了至少保证一次事件将被处理，我正在研究 akka-persistence。应用程序的其他部分已经使用 akka 和 kafka。

我的目标解决方案应该包含一个持久映射，可以通过它们的 eventId 轻松地从中选择事件。顺序不太重要。处理完事件后，可以将其从地图中移除（并且不应再保留）。

在找到的文档/示例中，我找到了满足每个事件清除要求的队列示例，但在简单查找方面遇到了困难（必须循环队列才能找到事件）。为了满足我想使用地图的简单查找，使用 PersistentActor 特征和下面的一些数据库。然而，事件由序列号清除（这将删除需要更多处理/正在等待其他事件发生的事件）。调查的另一个特征是 AtLeastOnceDelivery，其交付确认满足要求，但在处理完所有事件之前，此特征会阻止恢复。

关于如何在 Akka 中为事件实现持久性篮子的任何想法？（我正在使用 scala 顺便说一句）

Answer 1

这样的东西会满足您的需求吗？这可能不完全是你的逻辑，但基本上它接收一个新事件，坚持它接收到事件的事实，然后使用 id 将事件保存到地图中。然后在某个时候（不确定如何触发事件处理），它会收到处理具有特定 ID 的事件的命令。它坚持它应该处理事件的事实，然后处理事件并将其从地图中删除。这样，地图将在重新启动时恢复，并且您可以通过 Id 访问所有尚未处理的事件。

class PersistentMapActor extends PersistentActor {

    private var eventMap: Map[ Int, Event ] = Map[ Int, Event ]( )

    override def receiveRecover: Receive = {
            case NewEventSaved( payload: Event ) =>
                    eventMap = eventMap + ( (payload.eventId, payload) )
            case EventHandled( eventId ) =>
                    eventMap = eventMap - eventId
    }

    override def receiveCommand: Receive = {
            case SaveNewEvent( payload ) =>
                    persist( NewEventSaved( payload ) ) { persistedNewEvent =>
                            //Add new event to map
                            eventMap = eventMap + ( (payload.eventId, payload) )
                    }
            case HandleEvent( eventId ) =>
                    val event = eventMap.get( eventId )

                    event.foreach { e =>
                            persist( EventHandled( eventId ) ) { persistedEvent =>
                                    //Do stuff with the event
                                    //Remove the event from the map
                                    eventMap = eventMap - eventId
                            }
                    }
    }

    override def persistenceId: String = "PersistentMapActor"
}

object PersistentMapActor {

    case class Event( eventId: Int, someField: String )

    case class SaveNewEvent( payload: Event )

    case class NewEventSaved( payload: Event )

    case class HandleEvent( eventId: Int )

    case class EventHandled( eventId: Int )

}