使用Spring云流的Azure服务总线重试策略

lc8prwob  于 2023-02-08  发布在  Spring
关注(0)|答案(1)|浏览(153)

我正在使用Spring Cloud Stream和Azure Service Bus。以下是我的相关版本和依赖项：

<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}

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

spring-boot

1条答案

按热度按时间
nukf8bse

nukf8bse1#

要设置Service Bus绑定器的重试策略，需要在绑定级别中配置，例如，

spring:
  cloud:
    stream:
      servicebus:
        bindings:
          onReceive-in-0:
            consumer:
              retry:
                fixed:
                  delay: 100
                  max-retries: 5
                mode: fixed
赞(0）分享  回复(0）举报  2023-02-08
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com