访问投递尝试
要访问阻塞和非阻塞投递尝试次数,请将这些头添加到 @KafkaListener
方法签名中
@Header(KafkaHeaders.DELIVERY_ATTEMPT) int blockingAttempts,
@Header(name = RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS, required = false) Integer nonBlockingAttempts
只有将 ContainerProperties
的 deliveryAttemptHeader 设置为 true
时,才会提供阻塞投递尝试次数。
请注意,对于首次投递,非阻塞尝试次数将为 null
。
从 3.0.10 版本开始,提供了便捷的 KafkaMessageHeaderAccessor
,以便更简单地访问这些头;该访问器可以作为监听器方法的参数提供
@RetryableTopic(backoff = @Backoff(...)) @KafkaListener(id = "dh1", topics = "dh1") void listen(Thing thing, KafkaMessageHeaderAccessor accessor) { ... }
使用 accessor.getBlockingRetryDeliveryAttempt()
和 accessor.getNonBlockingRetryDeliveryAttempt()
获取值。如果未启用阻塞重试,该访问器将抛出 IllegalStateException
;对于非阻塞重试,该访问器在首次投递时返回 1
。