Google PubSub resent messages aren#39;t being processed(Google PubSub重新发送的消息未被处理)
本文介绍了Google PubSub重新发送的消息未被处理的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
我使用了Google PubSub文档中的订阅者示例 我所做的唯一修改是注释掉了对消息的确认。订阅者不再向队列中添加消息,而应根据Google云控制台中设置的间隔重新发送消息。
为什么会发生这种情况,还是我遗漏了什么?
public class SubscriberExample {
use the default project id
private static final String PROJECT_ID = ServiceOptions.getDefaultProjectId();
private static final BlockingQueue<PubsubMessage> messages = new LinkedBlockingDeque<>();
static class MessageReceiverExample implements MessageReceiver {
@Override
public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
messages.offer(message);
//consumer.ack();
}
}
/** Receive messages over a subscription. */
public static void main(String[] args) throws Exception {
// set subscriber id, eg. my-sub
String subscriptionId = args[0];
ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of(
PROJECT_ID, subscriptionId);
Subscriber subscriber = null;
try {
// create a subscriber bound to the asynchronous message receiver
subscriber = Subscriber.newBuilder(subscriptionName, new MessageReceiverExample()).build();
subscriber.startAsync().awaitRunning();
// Continue to listen to messages
while (true) {
PubsubMessage message = messages.take();
System.out.println("Message Id: " + message.getMessageId());
System.out.println("Data: " + message.getData().toStringUtf8());
}
} finally {
if (subscriber != null) {
subscriber.stopAsync();
}
}
}
}
JAVA
当您不确认消息时,推荐答案客户端库将对该消息调用modifyAckDeadline,直到maxAckExtensionPeriod传递。默认情况下,此值为一小时。因此,如果您不对消息进行确认/取消确认或更改此值,则消息很可能在一小时内不会重新传递。如果要更改最大确认扩展期限,请在构建器上进行设置:
subscriber = Subscriber.newBuilder(subscriptionName, new MessageReceiverExample())
.setMaxAckExtensionPeriod(Duration.ofSeconds(60))
.build();
还值得注意的是,如果不对消息进行ACK或NACK,则flow control可能会阻止更多消息的传递。默认情况下,Java客户端库允许最多1000条消息未完成,即等待ACK或NACK或等待最大ACK扩展期限过去。
这篇关于Google PubSub重新发送的消息未被处理的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持编程学习网!
沃梦达教程
本文标题为:Google PubSub重新发送的消息未被处理


猜你喜欢
- C++ 和 Java 进程之间的共享内存 2022-01-01
- Safepoint+stats 日志,输出 JDK12 中没有 vmop 操作 2022-01-01
- 如何使用WebFilter实现授权头检查 2022-01-01
- Jersey REST 客户端:发布多部分数据 2022-01-01
- Eclipse 插件更新错误日志在哪里? 2022-01-01
- Java包名称中单词分隔符的约定是什么? 2022-01-01
- Spring Boot连接到使用仲裁器运行的MongoDB副本集 2022-01-01
- 将log4j 1.2配置转换为log4j 2配置 2022-01-01
- value & 是什么意思?0xff 在 Java 中做什么? 2022-01-01
- 从 finally 块返回时 Java 的奇怪行为 2022-01-01