Akka持续性和至少一次输送

我有一些消息应用程序需要至少一次保证。我从文档中了解到：akka--持久性--是关于行动者状态的。它使用一些逐层的意识形态来处理这种状态。
在文档中，我发现了一些AbstractPersistentActorWithAtLeastOnceDelivery，它似乎提出了这种保证。但是现在我对一般的概念有一些怀疑：
我的演员是一个简单的传输器，我所需要的只是交付保证。所以我实际上并不关心演员的状态和所有消耗宝贵内存的分层。大量的日志会是问题的原因吗？
演员：

public class SafeSenderActor extends AbstractPersistentActorWithAtLeastOnceDelivery {

    private String persistenceId;
    private ActorSelection destination;

    public SafeSenderActor() {
        System.out.println("SafeSenderActor created");
        this.persistenceId = "safe-persistent-actor-id-" + UUID.randomUUID();
        destination = context().actorSelection("/user/safeReceiverRouter");
    }

    @Override
    public String persistenceId() {
        return persistenceId;
    }

    @Override
    public AbstractActor.Receive createReceive() {
        return ReceiveBuilder.create()
                .match(SenderTaskMessage.class, msg -> {
                    persistAsync(new MsgSentEvent(msg.getTestMessage()), this::updateState);
                })
                .match(ConfirmRobustMessageDelivery.class, ack -> {
                    persistAsync(new MsgConfirmEvent(ack.getMessageId(), ack.getLocalMessageNumber()), this::updateState);
                })
                .build();
    }

    @Override
    public Receive createReceiveRecover() {
        return receiveBuilder().match(Object.class, this::updateState).build();
    }

    private void updateState(Object event) {
        if (event instanceof MsgSentEvent) {
            MsgSentEvent ev = (MsgSentEvent) event;
            deliver(destination, deliveryId -> new RobustTestMessage(deliveryId, ev.getMessage()));
        } else if (event instanceof MsgConfirmEvent) {
            MsgConfirmEvent ev = (MsgConfirmEvent) event;
            confirmDelivery(ev.getDeliveryId());
        }
    }

}

在一些连续的消息之后，我收到错误：

[ERROR] [07/06/2017 01:40:33.446] [sender-system-akka.actor.default-dispatcher-50] [akka://sender-system@127.0.0.1:6666/user/safeSendersRouter/$d] Failed to persist event type [com.test.common.events.MsgSentEvent] with sequence number [358698] for persistenceId [safe-persistent-actor-id-648ec66d-7b7f-4291-b3a2-9bd395d92dc7]. (akka.pattern.CircuitBreaker$$anon$1: Circuit Breaker Timed out.)

我用leveldb作为日志。

这可能是一个很长的时间，但也许它仍然是有用的回答。您可以覆盖方法onPersistFailure和onPersistRejected，以调查您提到的错误。onPersistFailure方法被调用，如果持久失败。演员将停止。最好的做法是在一段时间后再次启动演员，并使用回退管理员。如果JOURNAl无法保存事件，则调用onPersistRejected方法。参与者为RESUMED。

public class SafeSenderActor extends AbstractPersistentActorWithAtLeastOnceDelivery implements ActorLogging {
    ...
    ...
    ...
    @Override
    public void onPersistFailure(Throwable cause, Object event, long seqNr) {
        log().error("fail to persist $event because of: {}", cause);
        super.onPersistFailure(cause, event, seqNr);
    }

    @Override
    public void onPersistRejected(Throwable cause, Object event, long seqNr) {
        log().error("persist rejected for {} because of: {}", event, cause);
        super.onPersistRejected(cause, event, seqNr);
    }
}

Akka持续性和至少一次输送

1条答案

相关问题

热门标签

最新问答