为什么添加 Thread.sleep 会使我的 Akka TestKit 单元测试通过？

Question

Java 8 和 Akka (Java API) 2.12:2.5.16 在这里。我有以下消息：

public class SomeMessage {
    private int anotherNum;

    public SomeMessage(int anotherNum) {
        this.anotherNum = anotherNum;
    }

    public int getAnotherNum() {
        return anotherNum;
    }

    public void setAnotherNum(int anotherNum) {
        this.anotherNum = anotherNum;
    }
}

还有以下演员：

public class TestActor extends AbstractActor {
    private Integer number;

    public TestActor(Integer number) {
        this.number = number;
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
            .matchAny(message -> {
                if (message instanceof SomeMessage) {
                    SomeMessage someMessage = (SomeMessage) message;
                    System.out.println("someMessage contains = " + someMessage.getAnotherNum());
                    someMessage.setAnotherNum(number);
                }
            }).build();
    }
}

还有下面的单元测试：

@RunWith(MockitoJUnitRunner.class)
public class TestActorTest {
    static ActorSystem actorSystem;

    @BeforeClass
    public static void setup() {
        actorSystem = ActorSystem.create();
    }

    @AfterClass
    public static void teardown() {
        TestKit.shutdownActorSystem(actorSystem, Duration.create("10 seconds"), true);
        actorSystem = null;
    }

    @Test
    public void should_alter_some_message() {
        // given
        ActorRef testActor = actorSystem.actorOf(Props.create(TestActor.class, 10), "test.actor");
        SomeMessage someMessage = new SomeMessage(5);

        // when
        testActor.tell(someMessage, ActorRef.noSender());

        // then
        assertEquals(10, someMessage.getAnotherNum());
    }
}

所以我要验证的是TestActor 确实收到了SomeMessage 并且它改变了它的内部状态。

当我运行这个单元测试时，它失败了，好像演员从来没有收到过消息：

java.lang.AssertionError: 
Expected :10
Actual   :5
 <Click to see difference>

    at org.junit.Assert.fail(Assert.java:88)
    at org.junit.Assert.failNotEquals(Assert.java:834)
    at org.junit.Assert.assertEquals(Assert.java:645)
  <rest of trace omitted for brevity>

[INFO] [01/30/2019 12:50:26.780] [default-akka.actor.default-dispatcher-2] [akka://default/user/test.actor] Message [myapp.actors.core.SomeMessage] without sender to Actor[akka://default/user/test.actor#2008219661] was not delivered. [1] dead letters encountered. If this is not an expected behavior, then [Actor[akka://default/user/test.actor#2008219661]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

但是当我修改测试方法并将Thread.sleep(5000) 引入其中时（在tell(...) 之后），它以出色的成绩通过：

@Test
public void should_alter_some_message() throws InterruptedException {
    // given
    ActorRef testActor = actorSystem.actorOf(Props.create(TestActor.class, null, 10), "test.actor");
    SomeMessage someMessage = new SomeMessage(5);

    // when
    testActor.tell(someMessage, ActorRef.noSender());

    Thread.sleep(5000);

    // then
    assertEquals(10, someMessage.getAnotherNum());
}

这是怎么回事？！ 显然我不希望我的演员测试到处都是sleeps，那么我在这里做错了什么，解决方法是什么？提前致谢！

Answer 1

@Asier Aranbarri 说你不让你的演员完成它的工作是正确的。

Actor 具有异步特性，尽管它们没有实现 Runnable，但它们与用于发送消息的线程分开执行。

您向参与者发送消息，然后立即断言该消息已更改。由于actor在异步上下文中运行，即不同的线程，它仍然没有处理传入的消息。因此，放置Threed.sleep 允许参与者处理消息，并且只有在此之后断言完成。

我可能会建议对您的初始设计进行一些更改，以使其与 akka 自然相结合。

首先，akka 不建议使用具有可变性的消息。它们必须是不可变的。在你的情况下，这被方法 SomeMessage#setAnotherNum 打破了。删除它：

public class SomeMessage {
    private int anotherNum;

    public SomeMessage(int anotherNum) {
        this.anotherNum = anotherNum;
    }

    public int getAnotherNum() {
        return anotherNum;
    }
}

在此之后创建SomeMessage 的新实例，而不是更改TestActor 中的传入消息并将其发送回context.sender()。就像这里定义的一样

static public class TestActor extends AbstractActor {
    private Integer number;

    public TestActor(Integer number) {
        this.number = number;
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .matchAny(message -> {
                    if (message instanceof SomeMessage) {
                        SomeMessage someMessage = (SomeMessage) message;
                        System.out.println("someMessage contains = " + someMessage.getAnotherNum());
                        context().sender().tell(new SomeMessage(number + someMessage.getAnotherNum()), context().self());
                    }
                }).build();
    }
}

现在，不是更改消息的内部状态，而是创建具有新状态的新消息，并将后面的消息返回给sender()。这是akka的正确用法。

这允许测试使用TestProbe 并重新定义如下

@Test
public void should_alter_some_message() {
    // given
    ActorRef testActor = actorSystem.actorOf(Props.create(TestActor.class,10));
    TestJavaActor.SomeMessage someMessage = new SomeMessage(5);
    TestProbe testProbe = TestProbe.apply(actorSystem);

    // when
    testActor.tell(someMessage, testProbe.ref());

    // then
    testProbe.expectMsg(new SomeMessage(15));
}

TestProbe 模拟发件人并捕获来自TestActor 的所有传入消息/回复。请注意，expectMsg(new SomeMessage(15)) 用于代替断言。它有一个内部阻塞机制，在断言完成之前等待消息被接收。这就是testing actors example 中发生的事情。

要使expectMsg 正确断言，您必须在您的类SomeMessage 中覆盖equals 方法

编辑：

为什么 Akka 在更改 SomeMessage 的内部状态时会皱眉？

akka 的强大功能之一是它不需要同步或等待/通知来控制对共享数据的访问。但这只能通过消息不变性来实现。想象一下，您发送了一条可变消息，您在 actor 处理它的确切时间更改了该消息。这可能会导致竞争条件。阅读this了解更多详情。

并且（2）同样适用于改变Actor的内部状态吗？ ActorRefs 具有可以更改的属性是否“可以”，或者社区是否也对此感到不满（如果是，为什么！）？

不，这不适用于这里。如果任何状态被封装在一个actor中并且只有它可以改变它，那么具有可变性是完全没问题的。

Answer 2

我认为你不要让演员做他的工作。也许AkkaActor 开始了他自己的线程？我认为Actor 确实实现了 Runnable，但我并不是 Akka 方面的专家。 --> 编辑 Actor 是一个界面，很高兴我说我不是专家..

我的猜测是，通过休眠你的主线程，你给了 Actor 的“线程”时间来完成他的方法。

我知道这可能没有帮助，但是太长了，无法发表评论。 : (