处理Akka actor有界邮箱MessageQueueAppendFailedException

发布于 2024-12-14 09:58:17 字数 978 浏览 0 评论 0原文

为了避免 OOM，我使用共享的自定义调度程序限制了一些 Akka 1.1.3 actor 的邮箱大小。例如：

object Static {
    val dispatcher = Dispatchers.newExecutorBasedEventDrivenWorkStealingDispatcher(
        "customDispatcher",
        1000,
        BoundedMailbox(capacity = 10)
    )
}

class MyActor extends Actor {
    self.dispatcher = Static.dispatcher
    ...
}

我想对邮箱溢出做出反应，以便我可以向上游生产者发送消息以暂停（旁注：遗憾的是，它看起来像 actor.stop()、等待和 actor .start() 将抛出一个 ActorStartException）。在队列填满和队列耗尽之间存在一些数据丢失是可以接受的。

Akka 关于调度程序的章节说

当尝试向 Actor 发送消息时，它会抛出 MessageQueueAppendFailedException(“BlockingMessageTransferQueue 如果消息无法添加到邮箱，则传输超时”）在pushTimeout指定的时间内。

我在哪里可以捕获这个异常？

该文档听起来像是我需要包装每个 myActor ！ try/catch 中的消息。是这样吗？我真的很想集中处理它。我的 Supervisor 可以拦截它并运行我的处理程序吗？

原文

To avoid OOM, I'm bounding the mailbox size of some of my Akka 1.1.3 actors with a shared custom dispatcher. For example:

object Static {
    val dispatcher = Dispatchers.newExecutorBasedEventDrivenWorkStealingDispatcher(
        "customDispatcher",
        1000,
        BoundedMailbox(capacity = 10)
    )
}

class MyActor extends Actor {
    self.dispatcher = Static.dispatcher
    ...
}

I'd like to react to the mailbox overflowing so I can message the upstream producers to pause (side note: it sadly looks like actor.stop(), wait, and actor.start() will throw an ActorStartException). Some data loss is acceptable between the queue filling up and the queue draining a bit.

Akka's chapter on Dispatchers says

When trying to send a message to the Actor it will throw a
MessageQueueAppendFailedException(“BlockingMessageTransferQueue
transfer timed out”) if the message cannot be added to the mailbox
within the time specified by the pushTimeout.

Where can I catch this exception?

The documentation makes it sound like I need to wrap every myActor ! message in a try/catch. Is that right? I'd really like to centralize its handling. Can my Supervisor perhaps intercept it and run my handler?

分享到QQ

分享到微博