java - 从 Actor 运行 CompletableFuture

Question

我正在使用 Akka (2.5.1) 在 Java (8) Spring Boot (1.5.2.RELEASE) 应用程序中尝试 react 模式。进展顺利，但现在我被困在尝试从 Actor 那里运行 CompletableFuture。为了模拟这个，我创建了一个非常简单的服务，它返回一个 CompletableFuture。但是，当我尝试将结果返回到调用 Controller 时，我收到有关死信的错误，并且没有返回任何响应。

我收到的错误是:

[INFO] [05/05/2017 13:12:25.650] [akka-spring-demo-akka.actor.default-dispatcher-5] [akka://akka-spring-demo/deadLetters] Message [java.lang.String] from Actor[akka://akka-spring-demo/user/$a#-1561144664] to Actor[akka://akka-spring-demo/deadLetters] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

这是我的代码。这是调用参与者的 Controller :

@Component
@Produces(MediaType.TEXT_PLAIN)
@Path("/")
public class AsyncController {
    @Autowired
    private ActorSystem system;

    private ActorRef getGreetingActorRef() {
        ActorRef greeter = system.actorOf(SPRING_EXTENSION_PROVIDER.get(system)
                  .props("greetingActor"));

        return greeter;
    }

    @GET
    @Path("/foo")
    public void test(@Suspended AsyncResponse asyncResponse, @QueryParam("echo") String echo) {
        ask(getGreetingActorRef(), new Greet(echo), 1000)
            .thenApply((greet) -> asyncResponse.resume(Response.ok(greet).build()));
    }
}

这是服务:

@Component
public class GreetingService {
    public CompletableFuture<String> greetAsync(String name) {
        return CompletableFuture.supplyAsync(() -> "Hello, " + name);
    }
}

然后这是接到电话的 Actor 。一开始我有这个:

@Component
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class GreetingActor extends AbstractActor {
    @Autowired
    private GreetingService greetingService;

    @Autowired
    private ActorSystem system;

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(Greet.class, this::onGreet)
                .build();
    }

    private void onGreet(Greet greet) {
        greetingService.greetAsync(greet.getMessage())
            .thenAccept((greetingResponse) -> getSender().tell(greetingResponse, getSelf()));
    }

}

这导致 2 个调用得到正确处理，但之后我会收到死信错误。然后我在这里读到可能导致我的问题的原因: http://doc.akka.io/docs/akka/2.5.1/java/actors.html

Warning When using future callbacks, inside actors you need to carefully avoid closing over the containing actor’s reference, i.e. do not call methods or access mutable state on the enclosing actor from within the callback. This would break the actor encapsulation and may introduce synchronization bugs and race conditions because the callback will be scheduled concurrently to the enclosing actor. Unfortunately there is not yet a way to detect these illegal accesses at compile time. See also: Actors and shared mutable state

所以我认为你的想法是将结果传递给 self()，然后你可以执行 getSender().tell(response, getSelf())。

所以我将代码更改为:

@Component
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class GreetingActor extends AbstractActor {
    @Autowired
    private GreetingService greetingService;

    @Autowired
    private ActorSystem system;

    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(Greet.class, this::onGreet)
                .match(String.class, this::onGreetingCompleted)
                .build();
    }

    private void onGreet(Greet greet) {
        pipe(greetingService.greetAsync(greet.getMessage()), system.dispatcher()).to(getSelf());
    }

    private void onGreetingCompleted(String greetingResponse) {
        getSender().tell(greetingResponse, getSelf());
    }

}

正在使用来自 GreetingService 的响应调用 onGreetingCompleted 方法，但当时我再次收到死信错误，因此由于某种原因它无法将响应发送回调用 Controller 。

请注意，如果我将服务更改为:

@Component
public class GreetingService {
    public String greet(String name) {
        return "Hello, " + name;
    }
}

Actor 中的 onGreet 为:

private void onGreet(Greet greet) {
    getSender().tell(greetingService.greet(greet.getMessage()), getSelf());
}

然后一切正常。因此，看来我的基本 Java/Spring/Akka 设置正确，只是当尝试从我的 Actor 调用 CompletableFuture 时，问题就开始了。

任何帮助将不胜感激，谢谢!

Answer 1

getSender 方法仅在消息同步处理期间可靠地返回发送者的引用。

在第一种情况下，您有:

 greetingService.greetAsync(greet.getMessage())
        .thenAccept((greetingResponse) -> getSender().tell(greetingResponse, getSelf()));

这意味着一旦 future 完成， getSender() 就会被异步调用。不再可靠了。您可以将其更改为:

 ActorRef sender = getSender();
 greetingService.greetAsync(greet.getMessage())
        .thenAccept((greetingResponse) -> sender.tell(greetingResponse, getSelf()));

<小时/>

在第二个示例中，您有

pipe(greetingService.greetAsync(greet.getMessage()), system.dispatcher()).to(getSelf());

您正在将响应传递给“getSelf()”，即您的 worker Actor 。原始发件人永远不会收到任何东西(因此询问过期)。您可以将其修复为:

pipe(greetingService.greetAsync(greet.getMessage()), system.dispatcher()).to(getSender());

<小时/>

在第三种情况下，在处理消息期间同步执行 getSender()，因此它可以工作。