Akka Camel - JMS 消息丢失 - 应该等待 Camel 的初始化吗? [英] Akka Camel - JMS messages lost - should wait for initialization of Camel?

问题描述

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

My experimental application is quite simple, trying what can be done with Actors and Akka.

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

After JVM start, it creates actor system with couple of plain actors, JMS consumer (akka.camel.Consumer) and JMS producer (akka.camel.Producer). It sends couple of messages between actors and also JMS producer -> JMS server -> JMS consumer. It basically talks to itself via JMS service.

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

From time to time I was experiencing weird behaviour: it seemed that from time to time, first of messages which where supposed to be sent to JMS server was somehow lost. By looking at my application logs, I could see that applications is trying to send the message but it was never received by JMS server. (For each run I have to start JVM&Application again).

Akka Camel Documentation 提到有可能某些组件可能在开始时未完全初始化:某些 Camel 组件可能需要一段时间才能启动，在某些情况下，您可能想知道端点何时被激活并准备好使用."

Akka Camel Documentation mentions that it's possible that some components may not be fully initialized at the begining: "Some Camel components can take a while to startup, and in some cases you might want to know when the endpoints are activated and ready to be used."

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

I tried to implement following to wait for Camel initialization

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)

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

which seems to help with this issue. (Although, when removing this step now, I'm not able to recreate this issue any more... :/).

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

My question is whether this is correct way to ensure all components are fully initialized?

我应该使用

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

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

for each akka.camel.Producer and akka.camel.Consumer actor to be sure that everything is initialized properly?

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

Is that all I should to do, or something else should be done as well? Documentation is not clean on that and it's not easy to test as issue was happening only occasionaly...

推荐答案

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

You need to initialize the camel JMS component and also Producer before sending any messages.

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

这篇关于Akka Camel - JMS 消息丢失 - 应该等待 Camel 的初始化吗?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！