Java，Akka Actor和有界邮件箱

Question

Java，Akka Actor和有界邮件箱

4

我在application.conf中有以下配置：

bounded-mailbox {
  mailbox-type = "akka.dispatch.BoundedMailbox"
  mailbox-capacity = 100
  mailbox-push-timeout-time = 3s
}

akka {

    loggers = ["akka.event.slf4j.Slf4jLogger"]
    loglevel = INFO
    daemonic = on
}

这是我配置演员的方式

public class MyTestActor extends UntypedActor implements RequiresMessageQueue<BoundedMessageQueueSemantics>{


    @Override
    public void onReceive(Object message) throws Exception {
        if (message instanceof String){

             Thread.sleep(500);
             System.out.println("message = " + message);
        }
        else {
            System.out.println("Unknown Message " );
        }

    }
}

现在我来初始化这个演员：

myTestActor = myActorSystem.actorOf(Props.create(MyTestActor.class).withMailbox("bounded-mailbox"), "simple-actor");

在这之后，在我的代码中，我向这个actor发送了3000条消息。

  for (int i =0;i<3000;i++){
        myTestActor.tell(guestName, null);}

我希望看到的是我的队列已满，但我的消息每半秒钟都会在onReceive方法中打印，就像什么也没有发生一样。因此，我认为我的邮箱配置未应用。

我做错了什么？

更新：我创建了一个订阅死信事件的Actor。

deadLetterActor = myActorSystem.actorOf(Props.create(DeadLetterMonitor.class),"deadLetter-monitor");

我已经安装了用于队列监控的Kamon：

向actor发送3000条消息后，Kamin显示如下内容：

Actor: user/simple-actor 

MailBox size:
 Min: 100  
Avg.: 100.0 
Max: 101

Actor: system/deadLetterListener 

MailBox size:
 Min: 0  
Avg.: 0.0 
Max: 0

Actor: system/deadLetter-monitor 

MailBox size:
 Min: 0  
Avg.: 0.0 
Max: 0

- danny.lesnik

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Andriy Plokhotnyuk · Accepted Answer

默认情况下，Akka将溢出的消息丢弃到DeadLetters中，而Actor不会停止处理：https://github.com/akka/akka/blob/876b8045a1fdb9cdd880eeab8b1611aa976576f6/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala#L411 但是，在丢弃消息之前，发送线程将被阻塞在由mailbox-push-timeout-time配置的时间间隔上。尝试将其减少到1毫秒并查看以下测试是否通过：

import java.util.concurrent.atomic.AtomicInteger

import akka.actor._
import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory._
import org.specs2.mutable.Specification

class BoundedActorSpec extends Specification {
  args(sequential = true)

  def config: Config = load(parseString(
    """
    bounded-mailbox {
      mailbox-type = "akka.dispatch.BoundedMailbox"
      mailbox-capacity = 100
      mailbox-push-timeout-time = 1ms
    }
    """))

  val system = ActorSystem("system", config)

  "some messages should go to dead letters" in {
    system.eventStream.subscribe(system.actorOf(Props(classOf[DeadLetterMetricsActor])), classOf[DeadLetter])
    val myTestActor = system.actorOf(Props(classOf[MyTestActor]).withMailbox("bounded-mailbox"))
    for (i  <- 0 until 3000) {
      myTestActor.tell("guestName", null)
    }
    Thread.sleep(100)
    system.shutdown()
    system.awaitTermination()
    DeadLetterMetricsActor.deadLetterCount.get must be greaterThan(0)
  }
}

class MyTestActor extends Actor {
  def receive = {
    case message: String =>
      Thread.sleep(500)
      println("message = " + message);
    case _ => println("Unknown Message")
  }
}

object DeadLetterMetricsActor {
  val deadLetterCount = new AtomicInteger
}

class DeadLetterMetricsActor extends Actor {
  def receive = {
    case _: DeadLetter => DeadLetterMetricsActor.deadLetterCount.incrementAndGet()
  }
}