java - AKKA - Spring MVC 和服务 - 超时异常

Question

我在 Spring MVC Web 应用程序中有一个 Spring 服务，它调用 Actor 系统来计算值。当我在 webapp 上多次触发时，应用程序会启动 TimeoutException。仅完成第一次计算。

你能给我一些帮助吗？

谢谢

@Service
public class Service {

    public static final int processors = Runtime.getRuntime().availableProcessors();
    @Value("${Iterations}")
    long numberOfIterations;
    @Value("${constante}")
    double constante;

    ActorSystem system;
    ActorRef master;

    public Serice() {
        // Create an Akka system
        system = ActorSystem.create("ComputationSystem");

        // create the master
        master = system.actorOf(new Props(new UntypedActorFactory() {
            public UntypedActor create() {
                return new Master(constante);
            }
        }));
    }

    @PreDestroy
    public void cleanUp() throws Exception {
        system.shutdown();
    }

    @Override
    public double calculatePrice(double x, double y, double z,
            double ex) {


        // start the calculation
        Work work = new Work(numberOfIterations, x, y, z,
                ex);

        Timeout timeout = new Timeout(Duration.create(60, "seconds"));
        Future<Object> future = ask(master, work, timeout);

        double total = 0;
        try {
            total = (Double) Await.result(future,
                    timeout.duration());
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {

        }

        return total;
    }

}


public class Master extends UntypedActor {
    private final ActorRef workerRouter;
    private double total = 0;
    private int answerReceived = 0;
    private long nbPerThreads;
    private double x;
    private double constante;
    private ActorRef replayTo;

    public Master(final double constante) {
        workerRouter = this.getContext().actorOf(
                new Props(new UntypedActorFactory() {
                    public UntypedActor create() {
                        return new Worker(constante);
                    }
                }).withRouter(new RoundRobinRouter(Algo.processors)),
                "workerRouter");
        this.constante = constante;
    }

    public void onReceive(Object message) {
        if (message instanceof Work) {
            Work work = (Work) message;

            replayTo = getSender();
            nbPerThreads = work.nbIterations / Algo.processors;
            x = work.x / 360.0;

            // Modify the message to give the right to the workers
            work.nbIterations = nbPerThreads;
            work.x = x;

            for (int i = 0; i < Algo.processors; i++) {

                workerRouter.tell(work, getSelf());
            }
            return;
        }

        if (message instanceof Double) {


            Double result = (Double) message;
            total += result;
            if (++answerReceived == Algo.processors) {
                double meanOfPremiums = total / (nbPerThreads * Algo.processors); 

                double result = Math.exp(-constante * x) * meanOfPremiums;

                System.out.println("returning answer :" + message);
                // Return the answer
                replayTo.tell(result, getSelf());
            }
            return;
        }
        unhandled(message);
    }
}

Answer 1

将发件人存储在属性中是有问题的。如果另一条工作消息在最后一条消息之前到达，则该消息将被覆盖，并且您的消息将无法正确到达。我的建议是创建一个临时参与者来汇总结果并回复发件人。每次收到工作消息时，您都会创建此参与者，并将其传递给需要回复的发送者作为参数。当您将工作发送到工作路由器时，您只需将这个新参与者作为发送者传递即可。您的工作人员代码没有变化。

这个新参与者只会在其 onReceive 方法中保存当前用于处理 Double 消息的代码，并在发送后调用 context().system().stop(self())回复原始发件人。

此模式应该会引导您找到可行的解决方案。