RabbitMQ骆驼消费者-使用单个消息

Question

我有一个场景，我想“拉”RabbitMQ队列/主题的消息，并一次处理一个。特别是当消费者启动时，队列中已经有消息。我尝试了以下方法，但没有成功（这意味着，这些选项中的每一个都会读取队列，直到队列为空，或者直到另一个线程关闭上下文）。

1.第一次处理后立即停止路由

final CamelContext context = new DefaultCamelContext();
try {
    context.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            RouteDefinition route = from("rabbitmq:harley?queue=IN&declare=false&autoDelete=false&hostname=localhost&portNumber=5672");
            route.process(new Processor() {
                Thread stopThread;

                @Override
                public void process(final Exchange exchange) throws Exception {
                    String name = exchange.getIn().getHeader(Exchange.FILE_NAME_ONLY, String.class);
                    String body = exchange.getIn().getBody(String.class);

                    // Doo some stuff
                    
                    routeComplete[0] = true;
                    if (stopThread == null) {
                        stopThread = new Thread() {
                            @Override
                        public void run() {
                                try {
                                    ((DefaultCamelContext)exchange.getContext()).stopRoute("RabbitRoute");
                                } catch (Exception e) {}
                            }
                        };
                    }
                    stopThread.start();
                }
            });
        }
    });
    context.start();
    while(!routeComplete[0].booleanValue())
        Thread.sleep(100);

    context.stop();
}

与1类似，但使用闩锁而不是while loop和sleep。

使用轮询消费者

 final CamelContext context = new DefaultCamelContext();
 context.start();
 Endpoint re = context.getEndpoint(srcRoute);
 re.start();
 try {
     PollingConsumer consumer = re.createPollingConsumer();
     consumer.start();
     Exchange exchange = consumer.receive();
     String bb = exchange.getIn().getBody(String.class);
     consumer.stop();
 } catch(Exception e){
     String mm = e.getMessage();
 }

使用ConsumerTemplate（）-类似于上面的代码。

我还尝试过启用预回迁，并将最大交换次数设置为1。

这些似乎都不起作用，如果队列中有3条消息，则在我能够停止路由之前，所有消息都已被读取。如果我要使用标准的RabbitMQ Java API，我会使用basicGet（）调用来读取单个消息，但出于其他原因，我更喜欢使用骆驼消费者。

是否有人能够成功地使用Camel RabbitMQ消费者处理包含多个消息的队列上的单个消息？

谢谢

Answer 1

这并不是组件的主要意图，因为它需要继续接收。但我已经创建了一个票证来研究如何支持basicGet（单一接收）。3.8版之后将推出一个新的基于spring的rabbitmq组件，因此它将在那里实现（首先）：https://issues.apache.org/jira/browse/CAMEL-16048