如何在侦听器使用骆驼接收它之前延迟队列中的 JMS 消息? [英] How to delay a JMS message in queue before a listener receives it using camel?

问题描述

我正在尝试将消息延迟几秒钟以使其进入队列.但是当我使用骆驼延迟选项时，它不是在队列中延迟，而是立即被消耗，并在路由路径中延迟.我们如何延迟消息以便它们在队列中等待几秒钟?

I am trying to delay a message to be in queue for few seconds. But when I use camel delay option, it is not delaying in queue, instead it is immediately consumed, and delaying in the route path. How can we delay messages so that they will be there in queue for few seconds?

我的骆驼配置弹簧如下所示.

My spring with camel configuration looks like below.

<bean id="jms" class="org.apache.camel.component.jms.JmsComponent">
    <property name="connectionFactory" ref="jmsConnectionFactory" />
</bean>

<camelContext id="camelContext" xmlns="http://camel.apache.org/schema/spring">  

    <route id="routeOne" delayer="10000">               
        <from uri="jms://queueone?concurrentConsumers=1"/>          
        <log message="routeOne incoming message ${body}"/>              
        <delay><constant>30000</constant></delay>                       
        <process ref="loggerProcessor"/>                        
    </route>

</camelContext> 

<bean id="loggerProcessor" name="loggerProcessor" class="emh.LoggerProcessor"/>

推荐答案

Delay via Camel

Camel delay 和 throttle 将从 ActiveMQ 队列中删除(消耗)消息并将其保留(在内存中)在路由中直到处理.事务 JMS 可能会缓解有关丢失消息的问题，尚未尝试过.值得深思.

Delay via Camel

Camel delay and throttle will remove (consume) the message from ActiveMQ queue and keep it (in-memory) in the route until processed. Transacted JMS might alleviate issues concerning losing the message, haven't tried that yet. Food for thought.

我使用 AMQ_SCHEDULED_DELAY 标头结合启用 schedulerSupport [1, 2, 3, 4、5、6].这是根据 JMS 2.0 交付延迟.

I got it to work in Spring Boot 2 with Camel and ActiveMQ using the AMQ_SCHEDULED_DELAY header combined with enabling schedulerSupport [1, 2, 3, 4, 5, 6]. This is per JMS 2.0 Delivery Delay.

请注意，只有当没有活动消费者时，消息才会出现在 ActiveMQ 队列中(pending) - 即应用程序关闭或 Camel 路由禁用.下面的配置用于使用现有 ActiveMQ 代理.

Note that a message will only appear in the ActiveMQ queue (pending) when there are no active consumers - i.e. application down or Camel route disabled. Configuration below is for using an existing ActiveMQ broker.

application.properties(来源)

spring.activemq.broker-url=tcp://localhost:61616

RouteBuilder

from("jms://incomingQueue").setHeader("AMQ_SCHEDULED_DELAY", constant("1000")).inOnly("jms://delayedQueue");  

from("jms://delayedQueue").process(...);

activemq.xml(现有代理安装)

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" schedulerSupport="true">  

对于在 /usr/local/Cellar/activemq//libexec/conf 下找到的 OSX Homebrew.

For OSX Homebrew found under /usr/local/Cellar/activemq/<version>/libexec/conf.

如果您想在您的应用程序中使用嵌入代理，您可以使用 BrokerService.

If you want to use a broker embedded in your application you can use the BrokerService.

默认情况下，数据保存在 activemq-data 中，并且需要 activemq-kahadb-store 依赖项 - 如果您选择 JDBC(来源).使用 brokerService.setPersistent( false ) 不再需要存储依赖项，但是 JMS 消息会再次存储在内存中.

By default data is persisted in activemq-data and requires the activemq-kahadb-store dependency - also if you chose JDBC (source). Using brokerService.setPersistent( false ) the store dependency is no longer required, but then the JMS message is stored in-memory again.

@Bean
public BrokerService brokerService()
{
    BrokerService brokerService = new BrokerService();
    brokerService.setSchedulerSupport( true );
    return brokerService;
}

可以找到更多示例 此处 和此处.

Further examples can be found here and here.

这篇关于如何在侦听器使用骆驼接收它之前延迟队列中的 JMS 消息?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！