Akka 和背压

Question

我看到了 Akka Streams 的背压概念，但我的项目没有使用 Akka Stream，我为 Akka 开发了另一个背压指示器概念，我想在这里问一下它是否可行...

我打算使用邮箱队列的深度作为背压的标准......背后的逻辑，如果 Akka 能够跟上我产生消息的速度，那么消息队列的深度应该肤浅。如果我发送的消息太多而 Akka 跟不上，那么消息队列中的消息将会越来越多，我必须降低生成消息的速度。

这与 Apache Cassandra 的“飞行中请求”基本相同......

Session.State state = session.getState();
for (Host host : state.getConnectedHosts()) {
   HostDistance distance = loadBalancingPolicy.distance(host);
   int connections = state.getOpenConnections(host);
   int inFlightQueries = state.getInFlightQueries(host);
   ....
}

所以我从 Akka 使用的 Akka reference.conf 中找到..

default-mailbox {
  mailbox-type = "akka.dispatch.SingleConsumerOnlyUnboundedMailbox"
}

class NodeMessageQueue extends AbstractNodeQueue[Envelope] with MessageQueue with 
  UnboundedMessageQueueSemantics {

 final def enqueue(receiver: ActorRef, handle: Envelope): Unit = add(handle)

 final def dequeue(): Envelope = poll()

所以我编写了一个 AspectJ Aspect 来拦截 enqueue() 和 dequeue()，并增加和减少一个全局 AtomicLong，我可以使用它来跟踪 Akka Actor 邮箱中等待的消息总数......

所以我的逻辑是，如果 Akka 能够跟上我发送的消息数量，那么该数量应该低于某个预先配置的值......

假设我有 100 000 个演员，他们在邮箱消息队列中有 1000 条消息，一切正常，但如果我看到 1 000 000 条消息在消息队列中等待，这是一个减慢消息生成的信号。 ...如果队列中有 10 000 000 条消息，则确定信号停止消息生产...

我构建了一个原型，它按我的预期工作......但在我继续之前，我想在这里问一下，你是否发现这个概念有任何真正的缺陷。

谢谢解答...

Answer 1

这可能是可行的，但是：

• 每条消息发送的单点争用（甚至是无锁的）在高容量下可能会对性能产生重大影响
• 您想要放慢速度的特定阈值在很大程度上取决于应用程序，并且可能会随着您更改应用程序而改变
• 根据系统的具体情况（以及“停止消息生成”的确切含义），可能会出现死锁（即，如果您要求不丢弃传入消息并且处理消息需要发送另一条消息（发送至少一条消息作为处理消息的一部分可能是 Akka 中最常见的事情），这意味着 Actor 将停止从其邮箱进行处理）。