Spring Cloud Stream异常处理过程解析

spring: cloud:  stream:   rocketmq:    binder:     name-server: 192.168.190.129:9876   bindings:    input:     destination: stream-test-topic     group: binder-group

package com.zj.node.usercenter.rocketmq;import lombok.extern.slf4j.Slf4j;import org.springframework.cloud.stream.annotation.StreamListener;import org.springframework.cloud.stream.messaging.Sink;import org.springframework.integration.annotation.ServiceActivator;import org.springframework.messaging.Message;import org.springframework.messaging.support.ErrorMessage;import org.springframework.stereotype.Service;/** * 消费者 * * @author 01 * @date 2019-08-10 **/@Slf4j@Servicepublic class TestStreamConsumer {  @StreamListener(Sink.INPUT)  public void receive1(String messageBody) {    log.info("消费消息，messageBody = {}", messageBody);    throw new IllegalArgumentException("参数错误");  }  /**   * 处理局部异常的方法   *   * @param errorMessage 异常消息对象   */  @ServiceActivator(    // 通过特定的格式指定处理哪个channel的异常    inputChannel = "stream-test-topic.binder-group.errors"  )  public void handleError(ErrorMessage errorMessage) {    // 获取异常对象    Throwable errorMessagePayload = errorMessage.getPayload();    log.error("发生异常", errorMessagePayload);    // 获取消息体    Message<?> originalMessage = errorMessage.getOriginalMessage();    if (originalMessage != null) {      log.error("消息体: {}", originalMessage.getPayload());    } else {      log.error("消息体为空");    }  }}

@StreamListener(Sink.INPUT)public void receive1(String messageBody) {  log.info("消费消息，messageBody = {}", messageBody);  throw new IllegalArgumentException("参数错误");}/** * 处理全局异常的方法 * * @param errorMessage 异常消息对象 */@StreamListener("errorChannel")public void handleError(ErrorMessage errorMessage) {  log.error("发生异常. errorMessage = {}", errorMessage);}

spring: cloud:  stream:   bindings:    input:     destination: stream-test-topic     group: binder-group   rabbit:    bindings:     input:      consumer:       # 自动将失败的消息发送给DLQ       auto-bind-dlq: true

spring: cloud:  stream:   rabbit:    bindings:     input:      consumer:       republish-to-dlq: true

# 默认是3，设为1则禁用重试spring.cloud.stream.bindings.<input channel名称>.consumer.max-attempts=1# 表示是否要requeue被拒绝的消息(即：requeue处理失败的消息)spring.cloud.stream.rabbit.bindings.input.consumer.requeue-rejected=true

spring: cloud:  stream:   bindings:    <input channel名称>:     consumer:      # 最多尝试处理几次，默认3      maxAttempts: 3      # 重试时初始避退间隔，单位毫秒，默认1000      backOffInitialInterval: 1000      # 重试时最大避退间隔，单位毫秒，默认10000      backOffMaxInterval: 10000      # 避退乘数，默认2.0      backOffMultiplier: 2.0      # 当listen抛出retryableExceptions未列出的异常时，是否要重试      defaultRetryable: true      # 异常是否允许重试的map映射      retryableExceptions:       java.lang.RuntimeException: true       java.lang.IllegalStateException: false

@Configurationclass RetryConfiguration {  @StreamRetryTemplate  public RetryTemplate sinkConsumerRetryTemplate() {    RetryTemplate retryTemplate = new RetryTemplate();    retryTemplate.setRetryPolicy(retryPolicy());    retryTemplate.setBackOffPolicy(backOffPolicy());    return retryTemplate;  }  private ExceptionClassifierRetryPolicy retryPolicy() {    BinaryExceptionClassifier keepRetryingClassifier = new BinaryExceptionClassifier(        Collections.singletonList(IllegalAccessException.class        ));    keepRetryingClassifier.setTraverseCauses(true);    SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(3);    AlwaysRetryPolicy alwaysRetryPolicy = new AlwaysRetryPolicy();    ExceptionClassifierRetryPolicy retryPolicy = new ExceptionClassifierRetryPolicy();    retryPolicy.setExceptionClassifier(        classifiable -> keepRetryingClassifier.classify(classifiable) ?            alwaysRetryPolicy : simpleRetryPolicy);    return retryPolicy;  }  private FixedBackOffPolicy backOffPolicy() {    final FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();    backOffPolicy.setBackOffPeriod(2);    return backOffPolicy;  }}

spring.cloud.stream.bindings.<input channel名称>.consumer.retry-template-name=myRetryTemplate