从一个演员开始运行一个完整的未来

Question

[信息][05/05/2017 13:12:25.650][Akka-Spring-demo-Akka.actor.default-dispatcher-5][Akka://Akka-spring-demo/deadletters]从演员[Akka://Akka-spring-demo/deadletters]到演员[Akka://Akka-spring-demo/deadletters]的消息[java.lang.string]没有传递。[1]遇到的一纸空文。可以通过配置设置'Akka.log-dead-letters'和'Akka.log-dead-letters-dhider-shutdown'关闭或调整此日志记录。

这是我的代码。这是调用参与者的控制器：

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

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

}

警告当使用将来的回调时，在actor内部，您需要小心地避免关闭包含actor的引用，即不要从回调中调用方法或访问封闭actor上的可变状态。这将破坏actor封装，并可能引入同步bug和争用条件，因为回调将同时调度到封装actor。不幸的是，还没有一种方法可以在编译时检测这些非法访问。另请参见：参与者和共享的可变状态

因此，我认为这个想法是将结果传输到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());
    }

}

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

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

那一切都很好。因此，我似乎已经正确地设置了基本的Java/Spring/Akka，只是当试图从我的演员中调用一个完整的未来时，问题就开始了。

如有任何帮助，不胜感激，谢谢！

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

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