java - 使用 Spring Cloud Stream 的 Azure 服务总线重试策略

Question

我正在将 Spring Cloud Stream 与 A​​zure 服务总线结合使用。这些是我的相关版本和依赖项:

 <spring-cloud-azure.version>4.3.0</spring-cloud-azure.version>
 <spring-cloud.version>2021.0.3</spring-cloud.version>

 <dependency>
     <groupId>com.azure.spring</groupId>
     <artifactId>spring-cloud-azure-starter</artifactId>
 </dependency>

 <dependency>
     <groupId>org.springframework.cloud</groupId>
     <artifactId>spring-cloud-stream</artifactId>
 </dependency>

 <dependency>
     <groupId>com.azure.spring</groupId>
     <artifactId>spring-cloud-azure-stream-binder-servicebus</artifactId>
     <version>4.3.0</version>
 </dependency>

我想实现重试策略，其中导致异常的消息会被重试多次，重试之间会有一些延迟。

从文档中，我看到有一些可用的配置设置，但似乎没有考虑这些设置。

配置属性:https://learn.microsoft.com/en-us/azure/developer/java/spring-framework/spring-cloud-azure-appendix

我的application.yml:

spring:
  cloud:
    stream:
      bindings:
        onReceive-in-0:
          destination: subscriber-1-input-queue

    azure:
      servicebus:
        consumer:
          retry:
            exponential:
              base-delay: 100
              max-delay: 1000
              max-retries: 5
            fixed:
              delay: 100
              max-retries: 5
            mode: exponential
        connection-string: Endpoint=...

我的消费者:

@Slf4j
@Component
public class ValuesConsumer {

    @Bean
    public Consumer<String> onReceive() {
        return (message) -> {

          log.info("Received the value {} in Consumer", message);

            switch (message) {
                case "IntegrationException" -> throw new IntegrationException();
                case "PoisonPillException" -> throw new PoisonPillException();
                case "ProcessingException" -> throw new ProcessingException();
                case "ValidationException" -> throw new ValidationException();
            }

        };
    }
}

当我抛出异常时，消息处理会重试 10 次，没有任何延迟 - 实际上不会发生重试逻辑，只是消息被放弃，然后再次从队列中接收。

这是异常堆栈跟踪，重复 10 次:

2022-08-31 11:26:45.819  INFO 10544 --- [oundedElastic-3] com.demoasb.ValuesConsumer        : Received the value IntegrationException in Consumer
2022-08-31 11:26:45.823 ERROR 10544 --- [oundedElastic-3] .s.i.s.i.ServiceBusInboundChannelAdapter : Error in the operation USER_CALLBACK occurred on entity subscriber-1-input-queue. Error: {}

com.azure.messaging.servicebus.ServiceBusException: Dispatcher has no subscribers for channel 'application.errorChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1@75a1e7f9]; nested exception is com.demoasb.exceptions.IntegrationException, failedMessage=GenericMessage [payload=byte[20], headers={azure_service_bus_expires_at=2022-09-14T09:26:45.740Z, azure_partition_key=415274637, azure_service_bus_received_message_context=com.azure.messaging.servicebus.ServiceBusReceivedMessageContext@6c625687, azure_service_bus_message_id=1981cbaf-68e3-62ec-b28e-9aaf5be48487, azure_service_bus_enqueued_sequence_number=46, azure_service_bus_enqueued_time=2022-08-31T09:26:45.740Z, azure_service_bus_lock_token=54cb3555-6253-4ef6-a858-de894358e56b, azure_service_bus_sequence_number=63, target-protocol=kafka, actionType=EventA, azure_service_bus_time_to_live=PT336H, azure_service_bus_delivery_count=0, azure_service_bus_state=ACTIVE, id=a896f227-28d2-c65e-afa3-b8e1e41b99c2, contentType=application/json, azure_service_bus_locked_until=2022-08-31T09:27:15.755Z, timestamp=1661938005819}], headers={id=d6a9c4de-a566-fb2d-f14f-568a3a2a4e43, timestamp=1661938005821}] for original GenericMessage [payload=byte[20], headers={azure_service_bus_expires_at=2022-09-14T09:26:45.740Z, azure_partition_key=415274637, azure_service_bus_received_message_context=com.azure.messaging.servicebus.ServiceBusReceivedMessageContext@6c625687, azure_service_bus_message_id=1981cbaf-68e3-62ec-b28e-9aaf5be48487, azure_service_bus_enqueued_sequence_number=46, azure_service_bus_enqueued_time=2022-08-31T09:26:45.740Z, azure_service_bus_lock_token=54cb3555-6253-4ef6-a858-de894358e56b, azure_service_bus_sequence_number=63, target-protocol=kafka, actionType=EventA, azure_service_bus_time_to_live=PT336H, azure_service_bus_delivery_count=0, azure_service_bus_state=ACTIVE, id=a896f227-28d2-c65e-afa3-b8e1e41b99c2, contentType=application/json, azure_service_bus_locked_until=2022-08-31T09:27:15.755Z, timestamp=1661938005819}]
    at com.azure.messaging.servicebus.ServiceBusProcessorClient$1.onNext(ServiceBusProcessorClient.java:346) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    at com.azure.messaging.servicebus.ServiceBusProcessorClient$1.onNext(ServiceBusProcessorClient.java:317) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java:440) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java:527) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.scheduler.ImmediateScheduler$ImmediateSchedulerWorker.schedule(ImmediateScheduler.java:84) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.trySchedule(FluxPublishOn.java:312) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.onNext(FluxPublishOn.java:237) ~[reactor-core-3.4.22.jar:3.4.22]
    at com.azure.messaging.servicebus.FluxAutoComplete$AutoCompleteSubscriber.hookOnNext(FluxAutoComplete.java:94) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    at com.azure.messaging.servicebus.FluxAutoComplete$AutoCompleteSubscriber.hookOnNext(FluxAutoComplete.java:55) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    at reactor.core.publisher.BaseSubscriber.onNext(BaseSubscriber.java:160) ~[reactor-core-3.4.22.jar:3.4.22]
    at com.azure.messaging.servicebus.FluxAutoLockRenew$LockRenewSubscriber.hookOnNext(FluxAutoLockRenew.java:178) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    at com.azure.messaging.servicebus.FluxAutoLockRenew$LockRenewSubscriber.hookOnNext(FluxAutoLockRenew.java:79) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    at reactor.core.publisher.BaseSubscriber.onNext(BaseSubscriber.java:160) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.4.22.jar:3.4.22]
    at com.azure.messaging.servicebus.implementation.ServiceBusReceiveLinkProcessor.drainQueue(ServiceBusReceiveLinkProcessor.java:516) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    at com.azure.messaging.servicebus.implementation.ServiceBusReceiveLinkProcessor.drain(ServiceBusReceiveLinkProcessor.java:475) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    at com.azure.messaging.servicebus.implementation.ServiceBusReceiveLinkProcessor.lambda$onNext$2(ServiceBusReceiveLinkProcessor.java:227) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java:440) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java:527) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) ~[reactor-core-3.4.22.jar:3.4.22]
    at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) ~[reactor-core-3.4.22.jar:3.4.22]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]
Caused by: org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.errorChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1@75a1e7f9]; nested exception is com.demoasb.exceptions.IntegrationException, failedMessage=GenericMessage [payload=byte[20], headers={azure_service_bus_expires_at=2022-09-14T09:26:45.740Z, azure_partition_key=415274637, azure_service_bus_received_message_context=com.azure.messaging.servicebus.ServiceBusReceivedMessageContext@6c625687, azure_service_bus_message_id=1981cbaf-68e3-62ec-b28e-9aaf5be48487, azure_service_bus_enqueued_sequence_number=46, azure_service_bus_enqueued_time=2022-08-31T09:26:45.740Z, azure_service_bus_lock_token=54cb3555-6253-4ef6-a858-de894358e56b, azure_service_bus_sequence_number=63, target-protocol=kafka, actionType=EventA, azure_service_bus_time_to_live=PT336H, azure_service_bus_delivery_count=0, azure_service_bus_state=ACTIVE, id=a896f227-28d2-c65e-afa3-b8e1e41b99c2, contentType=application/json, azure_service_bus_locked_until=2022-08-31T09:27:15.755Z, timestamp=1661938005819}], headers={id=d6a9c4de-a566-fb2d-f14f-568a3a2a4e43, timestamp=1661938005821}] for original GenericMessage [payload=byte[20], headers={azure_service_bus_expires_at=2022-09-14T09:26:45.740Z, azure_partition_key=415274637, azure_service_bus_received_message_context=com.azure.messaging.servicebus.ServiceBusReceivedMessageContext@6c625687, azure_service_bus_message_id=1981cbaf-68e3-62ec-b28e-9aaf5be48487, azure_service_bus_enqueued_sequence_number=46, azure_service_bus_enqueued_time=2022-08-31T09:26:45.740Z, azure_service_bus_lock_token=54cb3555-6253-4ef6-a858-de894358e56b, azure_service_bus_sequence_number=63, target-protocol=kafka, actionType=EventA, azure_service_bus_time_to_live=PT336H, azure_service_bus_delivery_count=0, azure_service_bus_state=ACTIVE, id=a896f227-28d2-c65e-afa3-b8e1e41b99c2, contentType=application/json, azure_service_bus_locked_until=2022-08-31T09:27:15.755Z, timestamp=1661938005819}]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:76) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187) ~[spring-messaging-5.3.22.jar:5.3.22]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) ~[spring-messaging-5.3.22.jar:5.3.22]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-5.3.22.jar:5.3.22]
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109) ~[spring-messaging-5.3.22.jar:5.3.22]
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:457) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.handler.AbstractMessageProducingHandler.doProduceOutput(AbstractMessageProducingHandler.java:325) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:268) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:232) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:142) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:56) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:222) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.dispatch(BroadcastingDispatcher.java:178) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187) ~[spring-messaging-5.3.22.jar:5.3.22]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166) ~[spring-messaging-5.3.22.jar:5.3.22]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-5.3.22.jar:5.3.22]
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109) ~[spring-messaging-5.3.22.jar:5.3.22]
    at org.springframework.integration.endpoint.MessageProducerSupport.sendErrorMessageIfNecessary(MessageProducerSupport.java:262) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:219) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at com.azure.spring.integration.servicebus.inbound.ServiceBusInboundChannelAdapter.access$600(ServiceBusInboundChannelAdapter.java:73) ~[spring-integration-azure-servicebus-4.3.0.jar:4.3.0]
    at com.azure.spring.integration.servicebus.inbound.ServiceBusInboundChannelAdapter$IntegrationRecordMessageListener.onMessage(ServiceBusInboundChannelAdapter.java:201) ~[spring-integration-azure-servicebus-4.3.0.jar:4.3.0]
    at com.azure.spring.integration.servicebus.inbound.ServiceBusInboundChannelAdapter$IntegrationRecordMessageListener.onMessage(ServiceBusInboundChannelAdapter.java:186) ~[spring-integration-azure-servicebus-4.3.0.jar:4.3.0]
    at com.azure.messaging.servicebus.ServiceBusProcessorClient$1.onNext(ServiceBusProcessorClient.java:343) ~[azure-messaging-servicebus-7.9.1.jar:7.9.1]
    ... 28 common frames omitted
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
    at org.springframework.integration.dispatcher.BroadcastingDispatcher.dispatch(BroadcastingDispatcher.java:153) ~[spring-integration-core-5.5.14.jar:5.5.14]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72) ~[spring-integration-core-5.5.14.jar:5.5.14]
    ... 55 common frames omitted

2022-08-31 11:26:45.824  WARN 10544 --- [oundedElastic-3] c.a.m.s.ServiceBusProcessorClient        : Error when processing message. Abandoning message.
Dispatcher has no subscribers for channel 'application.errorChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.function.FunctionConfiguration$FunctionToDestinationBinder$1@75a1e7f9]; nested exception is com.demoasb.exceptions.IntegrationException, failedMessage=GenericMessage [payload=byte[20], headers={azure_service_bus_expires_at=2022-09-14T09:26:45.740Z, azure_partition_key=415274637, azure_service_bus_received_message_context=com.azure.messaging.servicebus.ServiceBusReceivedMessageContext@6c625687, azure_service_bus_message_id=1981cbaf-68e3-62ec-b28e-9aaf5be48487, azure_service_bus_enqueued_sequence_number=46, azure_service_bus_enqueued_time=2022-08-31T09:26:45.740Z, azure_service_bus_lock_token=54cb3555-6253-4ef6-a858-de894358e56b, azure_service_bus_sequence_number=63, target-protocol=kafka, actionType=EventA, azure_service_bus_time_to_live=PT336H, azure_service_bus_delivery_count=0, azure_service_bus_state=ACTIVE, id=a896f227-28d2-c65e-afa3-b8e1e41b99c2, contentType=application/json, azure_service_bus_locked_until=2022-08-31T09:27:15.755Z, timestamp=1661938005819}], headers={id=d6a9c4de-a566-fb2d-f14f-568a3a2a4e43, timestamp=1661938005821}] for original GenericMessage [payload=byte[20], headers={azure_service_bus_expires_at=2022-09-14T09:26:45.740Z, azure_partition_key=415274637, azure_service_bus_received_message_context=com.azure.messaging.servicebus.ServiceBusReceivedMessageContext@6c625687, azure_service_bus_message_id=1981cbaf-68e3-62ec-b28e-9aaf5be48487, azure_service_bus_enqueued_sequence_number=46, azure_service_bus_enqueued_time=2022-08-31T09:26:45.740Z, azure_service_bus_lock_token=54cb3555-6253-4ef6-a858-de894358e56b, azure_service_bus_sequence_number=63, target-protocol=kafka, actionType=EventA, azure_service_bus_time_to_live=PT336H, azure_service_bus_delivery_count=0, azure_service_bus_state=ACTIVE, id=a896f227-28d2-c65e-afa3-b8e1e41b99c2, contentType=application/json, azure_service_bus_locked_until=2022-08-31T09:27:15.755Z, timestamp=1661938005819}]
2022-08-31 11:26:45.952  INFO 10544 --- [oundedElastic-3] c.a.m.s.i.ServiceBusReceiveLinkProcessor : {"az.sdk.message":"Adding credits.","prefetch":0,"requested":2,"linkCredits":0,"expectedTotalCredit":2,"queuedMessages":1,"creditsToAdd":1,"messageQueueSize":0}

如何正确实现延迟策略？我的配置中是否缺少任何内容？

Answer 1

要设置Service Bus Binder的重试策略，需要在绑定(bind)级别进行配置，例如:

spring:
  cloud:
    stream:
      servicebus:
        bindings:
          onReceive-in-0:
            consumer:
              retry:
                fixed:
                  delay: 100
                  max-retries: 5
                mode: fixed