如何在 Camel Route 中手动确认/确认 PubSub 消息 [英] How to manually ack/nack a PubSub message in Camel Route

问题描述

我正在使用 ackMode=NONE 设置骆驼路线，这意味着确认不会自动完成.如何明确确认路由中的消息?

I am setting up a Camel Route with ackMode=NONE meaning acknowlegements are not done automatically. How do I explicitly acknowledge the message in the route?

在我的 Camel Route 定义中，我将 ackMode 设置为 NONE.根据文档，我应该能够在下游手动确认消息:

In my Camel Route definition I've set ackMode to NONE. According to the documentation, I should be able to manually acknowledge the message downstream:

https://github.com/apache/camel/blob/master/components/camel-google-pubsub/src/main/docs/google-pubsub-component.adoc

AUTO = 交换在完成时得到 ack'ed/nack'ed.NONE = 下游进程必须明确地 ack/nack"

"AUTO = exchange gets ack’ed/nack’ed on completion. NONE = downstream process has to ack/nack explicitly"

但是我不知道如何发送确认.

However I cannot figure out how to send the ack.

from("google-pubsub:<project>:<subscription>?concurrentConsumers=1&maxMessagesPerPoll=1&ackMode=NONE")
                .bean("processingBean");

我的 PubSub 订阅的确认截止日期为 10 秒，因此由于 ackMode=NONE，我的消息每 10 秒重新发送一次.这正如预期的那样.但是，一旦处理完成并停止重新发送，我找不到手动确认消息的方法.

My PubSub subscription has an acknowledgement deadline of 10 seconds and so my message keeps getting re-sent every 10 seconds due to ackMode=NONE. This is as expected. However I cannot find a way to manually acknowledge the message once processing is complete and stop the re-deliveries.

推荐答案

我能够挖掘 Camel 组件并弄清楚它是如何完成的.首先我创建了一个 GooglePubSubConnectionFactory bean:

I was able to dig through the Camel components and figure out how it is done. First I created a GooglePubSubConnectionFactory bean:

@Bean
    public GooglePubsubConnectionFactory googlePubsubConnectionFactory() {
        GooglePubsubConnectionFactory connectionFactory = new GooglePubsubConnectionFactory();
        connectionFactory.setCredentialsFileLocation(pubsubKey);
        return connectionFactory;
    }

然后我可以从标头中引用消息的 ack id:

Then I was able to reference the ack id of the message from the header:

@Header(GooglePubsubConstants.ACK_ID) String ackId

然后我使用以下代码来确认消息:

Then I used the following code to acknowledge the message:

List<String > ackIdList = new ArrayList<>();
        ackIdList.add(ackId);
        AcknowledgeRequest ackRequest = new AcknowledgeRequest().setAckIds(ackIdList);
        Pubsub pubsub = googlePubsubConnectionFactory.getDefaultClient();
        pubsub.projects().subscriptions().acknowledge("projects/<my project>/subscriptions/<my subscription>", ackRequest).execute();

这篇关于如何在 Camel Route 中手动确认/确认 PubSub 消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！