如何使用Camel路由每4小时读取一次RabbitMQ队列消息？

算力阿峰

2026-6-24

解决方案

1. 修改Quartz的Cron表达式

每4小时执行一次的Cron表达式为 0 0 */4 * * ?，替换原有的2分钟触发表达式。该表达式的具体含义：

0：秒位，固定在0秒触发
0：分位，固定在0分触发
*/4：小时位，每间隔4小时触发一次
*：日期位，每天都触发
*：月份位，每个月都触发
?：星期位，不指定（避免与日期位规则冲突）

注：Camel配置中Cron表达式的空格需要用+替换，因此最终配置值为0+0+*/4+*+*+?

2. 修正路由逻辑

你原路由中使用to("spring-rabbitmq:...")是发送消息到RabbitMQ，并非拉取队列消息。要实现定时拉取队列内容，需使用pollEnrich组件主动拉取，修改后的Route代码如下：

public class DummyProcessorRoute extends RouteBuilder {

    @Autowired
    DummyQ3Processor dummyQ3Processor;

    @Override
    public void configure() throws Exception {
        // 每4小时触发定时任务
        from("quartz://myTimer?cron=0+0+*/4+*+*+?")
            // 拉取RabbitMQ队列消息，timeout=0表示无消息时立即返回
            .pollEnrich("spring-rabbitmq:{{rabbitmq.in.exchange}}?exchangeType=topic&queues={{in.hd.dummy}}", 0)
            // 仅当拉取到有效消息时才执行后续处理
            .filter(body().isNotNull())
            .process(dummyQ3Processor);
    }
}

3. 简化Processor代码

原Processor中exchange.getIn().setBody(messageBody);属于冗余操作，直接读取并打印即可：

@Service
public class DummyQ3Processor implements Processor  {

    @Override
    public void process(Exchange exchange) {
        String messageBody = exchange.getIn().getBody(String.class);
        System.out.println("Message body:" + messageBody);
    }
}