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

Question

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

JVM启动后，它创建了一个带有几个普通角色的角色系统，即JMS消费者(akka.camel.Consumer)和JMS生产者(akka.camel.Producer)。它在演员和JMS制作人之间传递一些信息-

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

Akka Camel文档提到，有些组件可能一开始就没有完全初始化：“有些Camel组件可能需要一段时间才能启动，在某些情况下，您可能想知道endpoint何时激活并准备好使用。”

我尝试实现以下操作以等待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。制作人和akka.camel。消费者参与者是否要确保一切都正确初始化？

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

Answer 1

在发送任何消息之前，您需要初始化骆驼JMS组件和生产者。

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