Я пытаюсь реализовать Saga (Process Manager) с PersistentActor, который обновляет несколько сущностей / совокупных корней для достижения (конечной) согласованности без транзакции ACID.
Скажем, в середине процесса, между получением сообщения диспетчером процессов и началом его обработки, произошел сбой jvm.
Или приложение в данный момент разворачивается заново.
Будет ли почтовый ящик диспетчера процессов восстановлен после перезапуска jvm (вместе с системой акторов)?
Почтовые ящики не восстанавливаются в случае сбоя JVM. Ответственность за то, чтобы его сообщение было получено / сохранялась, лежит на отправителе. Между постоянными участниками используйте At -Минимальная доставка.
Учитывая, что вы уже используете PersistentActor, разумно использовать его функции AtLeastOnceDelivery для того, что вам нужно. AtLeastOnceDelivery по существу гарантирует, что действующий субъект-отправитель будет продолжать отправлять сообщения получателю с настраиваемой частотой, пока он не получит подтверждение от получателя (или от некоторой процедуры завершения для отмены текущих попыток доставки).
Как и в случае с обычным PersistentActor (с AtLeastOnceDelivery или без него), сохраняемые сообщения будут повторно отправлены в случае сбоя JVM. В случае AtLeastOnceDelivery сохраненные сообщения будут повторно отправлены после восстановления системы до получения подтверждения. Akka предоставляет метод deliver для отправки каждого сообщения, помеченного последовательным монотонно возрастающим deliveryId, и, получив подтверждение с соответствующим deliveryId от получателя, использует метод confirmDelivery для сигнализации успешной доставки сообщения.
Следующий фрагмент, который является частью образца кода из соответствующего Akka doc, выделяет ключевую логику AtLeastOnceDelivery в классе актора-отправителя (который extends PersistentActor with AtLeastOnceDelivery). Обратите внимание, что один и тот же обработчик сохраняемости updateState используется для сохранения событий, соответствующих deliver и confirmDelivery, для согласованного восстановления состояния:
override def receiveCommand: Receive = {
case s: String =>
persist(MsgSent(s))(updateState)
case Confirm(deliveryId) =>
persist(MsgConfirmed(deliveryId))(updateState)
}
override def receiveRecover: Receive = {
case evt: Evt => updateState(evt)
}
def updateState(evt: Evt): Unit = evt match {
case MsgSent(s) =>
deliver(destination)(deliveryId => Msg(deliveryId, s))
case MsgConfirmed(deliveryId) =>
confirmDelivery(deliveryId)
}
