Akka Camel - JMS 消息丢失 - 应该等待 Camel 的初始化吗？

Question

我的实验应用程序非常简单，尝试使用 Actors 和 Akka 可以做什么。

在 JVM 启动后，它创建了带有几个普通 Actor 的 Actor 系统，JMS 消费者 (akka.camel.Consumer) 和 JMS 生产者 (akka.camel.Producer)。它在参与者和 JMS 生产者 -> JMS 服务器 -> JMS 消费者之间发送一些消息。它基本上通过 JMS 服务与自己对话。

我不时遇到奇怪的行为：似乎不时地，应该发送到 JMS 服务器的第一条消息不知何故丢失了。通过查看我的应用程序日志，我可以看到应用程序正在尝试发送消息，但 JMS 服务器从未收到它。 （对于每次运行，我都必须再次启动 JVM&Application）。

Akka Camel Documentation 提到某些组件可能在开始时未完全初始化：“某些 Camel 组件可能需要一段时间才能启动，在某些情况下您可能想知道端点何时被激活并且可以使用了。"

我尝试实现以下等待骆驼初始化

val system = ActorSystem("actor-system")
val camel = CamelExtension(system)

val jmsConsumer = system.actorOf(Props[JMSConsumer])
val activationFuture = camel.activationFutureFor(jmsConsumer)(timeout = 10 seconds, executor = system.dispatcher)
val result = Await.result(activationFuture,10 seconds)

这似乎有助于解决这个问题。 （虽然，现在删除此步骤时，我无法再重新创建此问题...：/）。

我的问题是，这是否是确保所有组件完全初始化的正确方法？

我应该使用

val future = camel.activationFutureFor(actor)(timeout = 10 seconds, executor = system.dispatcher)
Await.result(future, 10 seconds)

为每个 akka.camel.Producer 和 akka.camel.Consumer 演员确保一切都正确初始化？

这是我应该做的全部，还是应该做其他事情？文档不干净，也不容易测试，因为问题只是偶尔发生......

Answer 1

在发送任何消息之前，您需要初始化camel JMS组件和Producer。

import static java.util.concurrent.TimeUnit.SECONDS;

import scala.concurrent.Future;

import scala.concurrent.duration.Duration;

import akka.dispatch.OnComplete;

ActorRef producer = system.actorOf(new Props(SimpleProducer.class), "simpleproducer"); 
Timeout timeout = new Timeout(Duration.create(15, SECONDS));

Future<ActorRef> activationFuture = camel.activationFutureFor(producer,timeout,  system.dispatcher());

activationFuture.onComplete(new OnComplete<ActorRef>() {
            @Override
            public void onComplete(Throwable arg0, ActorRef arg1)
                    throws Throwable {

                producer.tell("First!!");
            }
            },system.dispatcher());