从前慢现在也慢

这个屌丝很懒，什么也没留下！

热门标签

事件来源Event sourcing

作者：从前慢现在也慢 | 2024-03-02 08:07:30

踩

event sourcing

一、事件来源基本思想：

当一个持久化actor接收到一个（非持久化）命令，首先它要验证（这个命令）是否可以运用到当前状态。

如果命令验证成功，根据这个命令产生一个事件。在事件成功持久化之后，可以用来改变actor的状态。

当持久化actor需要恢复时，因为之前已经验证过可以运用到当前状态，我们可以直接将持久化的事件进行重放。

Akka持久化通过 Abstract PersistentActor 支持事件来源。actor可以使用persist方法持久化和处理事件。通过实现 CreateReceiveRecover 和 CreateReceive 定义 AbstractPersistentActor 的行为。

CreateReceiveRecover 方法是在恢复过程中处理事件和快照消息。

CreateReceive 方法是用来处理普通的Actor的消息。若Actor收到命令的话会调用persist方法

persist方法是异步的方式持久化事件。它有两个参数，一个是事件，另一个是事件处理程序（event handler）。

事件处理程序是将之前持久化过的事件进行处理，该事件在内部作为独立消息发送回持久化actor 来使事件处理程序执行，来 改变或关闭持久化actor的状态。持久化事件的发送者也是相应命令的发送者，因此当命令的发送者没显出时事件处理程序也可以回复。

当使用persist方法来持久化事件时在调用和执行相关事件处理程序的过程中，要保证持久化actor不会收到下一步的命令，否则会受到影响。当在某个命令的上下文中多次调用 persist方法时。这个过程中收到的消息一直被暂存直到 presist方法运行结束。

如果实例化事件失败，onPersistFailure方法将被调用（默认记录为error），并且actor将无条件地被停止。如果持久化事件在存储之前被拒绝，比如事件发生连续错误，onPersistRejected将被调用（默认记录为warning）并且actor继续下一条消息

二、持久化

标识符

一个持久化actor必须有个标识符这个标识符必须用 persistenceId 方法来定义。

恢复

PersistentActor在启动和重启时通过重放之前持久化的日志消息来实现自动恢复，如果在恢复过程中收到新的消息，会将新消息先存储起来等恢复完成后，在收到新的消息。

可以限制同一时间并发的恢复的数量，来限制系统和后端的数据存储不超载，如果超过限制actor将等待到其他恢复都完成后才开始。配置方式：

akka.persistence.max-concurrent-recoveries = 50

自定义恢复

在应用程序中有时也需要依照客户具体要求来恢复，通过返回 recovery 方法中的自定义 Recovery对象来执行自定义恢复。 recovery 是 PersistentActor的一个方法。

你可以使用 SnapshotSelectionCriteria.None. 来跳过加载快照和重放所有事件。它用于将快照序列化格式变成互不相容的方式时。不适宜用于事件被删除的情况下。

@Overridepublic Recovery recovery() { return Recovery.create(SnapshotSelectionCriteria.none());}

另一个可能的自定义恢复是设置重放的上界，对debug很有帮助，使得actor仅在过去的某个点重放。

@Overridepublic Recovery recovery() { return Recovery.create(457L);}

在PersistentActor的recovery 方法中返回Recovery.none()可以使恢复失效。

@Overridepublic Recovery recovery() { return Recovery.none();}

恢复状态

持久化actor可以通过以下方法查询它自己的恢复状态

public boolean recoveryRunning();public boolean recoveryFinished();

持久化actor在回复完成后会收到一个特殊的RecoveryCompleted 消息。然后再执行下一步操作

如果actor从日志中的恢复状态有问题，onRecoveryFailure 会被调用（记录为error）并且actor将被停止。

三、内部暂存（stash）

持久化actor有一个私有的暂存用来缓存整个恢复过程中进来的消息或者暂存 persist\persistAll方法持久化的事件。内部暂存通过挂钩到 unstashAll 与普通暂存协作

你应该控制消息的产出不要超过持久化actor的处理能力，否则暂存消息的数量将无限增长。所以我们要在mailbox配置中定义暂存的容量来保护暂存并防止发生 OutOfMemoryError

akka.actor.default-mailbox.stash-capacity=10000

注意，如果你有很多持久化actor，要定义一个小的暂存容量，防止占用过多的内存

持久化actor定义了三个策略来处理内部暂存容量超出的故障。默认的溢出策略是 ThrowOverflowExceptionStrategy，具体内容是丢弃当前的信息，抛出 StashOverflowException异常，造成actor重启。

你可以覆盖 internalStashOverflowStrategy 方法为了“独特的”持久化actor来返回 DiscardToDeadLetterStrategy 或者 ReplyToStrategy 或者通过提供FQCN(Fully Qualified Class Name完全限定类名）来给所有的持久化actor来定义“默认值”。

在persistence 的配置中：

akka.persistence.internal-stash-overflow-strategy="akka.persistence.ThrowExceptionConfigurator"

DiscardToDeadLetterStrategy 策略也有一个打包好的配akka.persistence.DiscardConfigurator.

你也可以查询默认策略：

Persistence(context.system).defaultInternalStashOverflowStrategy

四、放宽的局部一致性要求和高吞吐量的用例

如果面临放宽的局部一致性要求和高吞吐量，有时 PersistentActor及其 persist在处理大量涌入的命令时可能会不够，有时你可能会放宽一致性要求——例如你会想要尽可能快速地处理命令，假设事件最终会持久化并在后台恰当处理，并在需要时追溯性地回应持久性故障。

persistAsync方法提供了一个工具，用于实现高吞吐量的持久化actor。当日志仍在致力于持久化和执行用户事件回调代码时，它 不会暂存传入的命令。

推迟操作，直到持久化处理程序已经执行

PersistentActor 提供了一个实用的方法 deferAsync(延迟异步)，它工作起来类似于 persistAsync但是不持久化传递过的事件，它将保留在内存中，并在调用处理程序时使用。建议将其用于读取操作，以及在domain模型中没有相应事件的操作。

请注意， sender()在处理程序回调中是安全的，将指向该命令的原始发送方，该命令将调用这个 deferAsync处理程序。

持久化嵌套调用

可以在各自的回调块中调用 persistAsync 和 persist，它们将适当地保留线程安全性(包括sender的正确值)和存储保证。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/179728