2

当我在 GCP 环境中测试应用程序时，我遇到了非常奇怪的浮动 bug()。我找不到具体的重现步骤，但它确实不时发生。

我看到该消息已成功确认：

2019-12-06 12:37:47.348  INFO 1 --- [sub-subscriber3] .i.g.MyAcknowledgementHandler : Acknowledged message - 1575635858865987

我有以下代码要确认：

        var generation = message.getHeaders().get("objectGeneration");
        pubSubMessage = message.getHeaders().get(GcpPubSubHeaders.ORIGINAL_MESSAGE, BasicAcknowledgeablePubsubMessage.class)
        pubSubMessage.ack().addCallback(
                v -> {
                    removeFromIdempotentStore(targetMessage, false);
                    log.info("Acknowledged message - {}", generation);
                },
                e -> {
                    removeFromIdempotentStore(targetMessage, false);
                    log.error("Failed to acknowledge message - {}", generation, e);
                }
        );

我还看到以下日志：

2019-12-06 12:37:48.868 WARN 1 --- [sub-subscriber1] c.b.m.i.MyDiscardedMessagesHandler : Duplicate message received GenericMessage [... headers={gcp_pubsub_acknowledgement=org.springframework.cloud.gcp.pubsub.integration.inbound.PubSubInboundChannelAdapter$1@1abafe68, bxwid=12345, errorChannel=org.springframework.messaging.core.GenericMessagingTemplate$TemporaryReplyChannel@3c3efd63, idempotent.keys=[objectId.mixed emails.csv, objectGeneration.1575635858865987].....

并且不断重复。此外，我在订阅图中看到消息仍然存在（在确认回调调用之后）

丢弃逻辑：

....
.gateway(nexrFlow, idempotentByHeader("objectId")); 


Consumer<GatewayEndpointSpec> idempotentByHeader(String objectIdHeader) {
    return endpointSpec -> endpointSpec.advice(idempotentByHeaderInterceptor(objectIdHeader))
            .errorChannel(errorChannel())
            .replyTimeout(0L);
}

default IdempotentReceiverInterceptor idempotentByHeaderInterceptor(String header) {
    MessageProcessor<String> headerSelector = message -> headerExpression(header).apply(message);
    var interceptor = new IdempotentReceiverInterceptor(new MetadataStoreSelector(headerSelector, idempotencyStore()));
    interceptor.setDiscardChannel(idempotentDiscardChannel());
    return interceptor;
}

我不知道如何解决它。有任何想法吗？

4

1 回答 1

3

Pub/sub 旨在保证至少一条消息的传递，因此这是预期的行为。查看产品常见问题以获取官方解释。

如那里所述，频繁重复的一个原因是消息未在确认期限内确认。如果该消息的处理时间超过最后期限，则重新发送该消息。这就是我在之前的评论中要求 AckDeadline 的原因。默认情况下应该是 10 秒。您可以检查您如何在控制台中配置它，单击您正在使用的订阅。您可以尝试增加它以等待更多消息完成处理。通过在订阅中单击一次编辑来执行此操作。

但是，即使您在截止日期内确认，有时也会发生重复。这对于保证至少一次交付是必要的。

于 2019-12-17T15:59:55.390 回答