java - 为什么添加 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;
    }
}

还有以下 Actor :

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 并且它改变了它的内部状态。

当我运行这个单元测试时，它失败了，就好像 Actor 从未收到消息一样:

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());
}

这是怎么回事？! 显然我不希望我的 actor 测试中充斥着sleep，那么我在这里做错了什么，解决方法是什么？提前致谢!

Answer 1

@Asier Aranbarri 说得对，您不让 Actor 完成工作。

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

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

我可能会建议对您的初始设计进行一些更改，以便与 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 方法

编辑:

Why does Akka frown upon changing SomeMessage's internal state?

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

And (2) does the same apply to changing Actors' internal state? Is it "OK" for ActorRefs to have properties that can be changed, or does the community frown upon that as well (and if so, why!)?

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