我似乎在 20 秒后收到超时错误。我有一个自定义处理器实现处理器。我注入了一个 DAO，当在自定义处理器中查找数据时，在 Apache Camel 端查找数据需要更长的时间并且它会超时。如果我在没有 Apache Camel 的情况下运行相同的代码，它会立即运行。通过在 CustomProcessor 中执行 SELECT，查找数据需要更长的时间。
DAO 的内存引用是相同的，因此在测试中立即获取数据，CustomProcessor 在收到数据之前挂起 20 秒并抛出异常。
我无法弄清楚问题的原因。
我在 Gihib 上找到了代码：https://github.com/rajivj2/example2
问题出在 StatusHibernateDAO 的第 27 行。我使用只有一个表的内存数据库。仅填充数据。
在不使用 Apache Camel 的情况下使用 CustomProcessor 时，它可以完美运行。
希望你能帮上忙。
【问题讨论】：
标签： java spring jpa apache-camel dao
由于您使用的是基于注释的配置，嵌入式 Active MQ 代理可能会意外停止，最好使用外部代理（例如：tcp://host:61616）。
JMS 端点上的 QoS 设置似乎不起作用。您可以从 ProducerTemplate 中获取 Header 值。
由于您使用的是 producerTemplate.requestBody，因此 activemq 端点假定它是请求和重播交换（InOut），并且由于路由超时没有响应。如果您想实施 (InOut)，请按照 http://camel.apache.org/jms.html 的说明进行操作。由于您的 Routebuilder 是 InOnly，您需要从 ProducerTemplate 发送 disableReplyTo 标头。
将您的测试方法 testMessageSendToConsumerQueueRemoteId 替换为
@Transactional
@Test
public void testMessageSendToConsumerQueueRemoteId() throws Exception {
Status status = new Status();
status.setUserId(10);
statusDAO.save(status);
Endpoint mockEndpoint = this.context.getEndpoint(properties.getProperty("activemq.destination"));
PollingConsumer consumer = mockEndpoint.createPollingConsumer();
producerTemplate.sendBodyAndHeader(source, "<?xml version='1.0' encoding='UTF-8' standalone='yes'?><example><remoteid>10</remoteid></example>","disableReplyTo","true");
Status savedStatus = consumer.receive(100).getIn().getBody(Status.class);
logger.info("savedStatus "+savedStatus.getID()+" "+savedStatus.getUserId());
assertNotNull(savedStatus);
}
将你的 ContentEnricherProcessor 的处理方法替换为
public void process(Exchange exchange) throws Exception {
Message message = exchange.getIn();
Entity entity = (Entity) message.getBody();
Status status = process(entity);
message.setBody(status);
exchange.getOut().setBody(status);
}
你的 camel.property 文件应该是
source=activemq:queue:deliverynotification
activemq.location=tcp://localhost:61616
activemq.destination=activemq:queue:responsenotification
如果您想收到生成的响应，您需要更改您的 RouteBuilder。
from(properties.getProperty("source"))
.process(new ResponseProcessor())
.inOnly("direct:z")
.end();
from("direct:z")
.unmarshal()
.jaxb("com.example.entities.xml").convertBodyTo(Entity.class)
.multicast()
.to("direct:x")
.end();
from("direct:x").transacted()
.process((ContentEnricherProcessor) applicationContext.getBean("contentEnricherProcessor"))
.to(properties.getProperty("activemq.destination"));
然后像这样改变生产者模式
Status response = producerTemplate.requestBody(source, "<?xml version='1.0' encoding='UTF-8' standalone='yes'?><example><remoteid>11</remoteid></example>",Status.class);
这里是ResponseProcessor的处理方法
public void process(Exchange exchange) throws Exception {
Message outMsg = exchange.getIn().copy();
exchange.setOut(outMsg);
}
Camel ROCKS...您可以实现几乎任何企业集成用例或模式:)
【讨论】：