Spring Cloud Stream 异常处理

r囧r小猫 2023-03-04 05:26 113阅读 0赞

### 文章目录 ###

*   *  Spring Cloud Stream 异常处理
     *   *  失败自动重试
         *  自定义错误处理逻辑
         *  全局异常处理
         *  死信队列
     *  参考

## Spring Cloud Stream 异常处理 ##

### 失败自动重试 ###

Spring Cloud Stream 支持失败之后自动重试，其实已经包含了默认的配置，但是我们可以通过如下的配置自定义我们的配置。

spring:
     rabbitmq:
        host: 192.168.99.100
        port: 5672
        username: guest
        password: guest
     cloud:
       stream:
         bindings:
           input:
             destination: greetings
             consumer:
               max-attempts: 4 # 默认为3
               backOffInitialInterval: 5000 #消息消费失败后重试消费消息的初始化间隔时间。默认1000ms，即第一次重试消费会在1s后进行
               backOffMultiplier: 5 #相邻两次重试之间的间隔时间的倍数。默认2，即第二次是第一次间隔时间的2倍，第三次是第二次的2倍
               backOffMaxInterval: 60000 #下一次尝试重试的最大时间间隔，默认为10000ms，即10s

代码参考：[https://gitee.com/cckevincyh/cloud-stream-demo/tree/demo11/][https_gitee.com_cckevincyh_cloud-stream-demo_tree_demo11]

### 自定义错误处理逻辑 ###

首先我们的配置如下：

自定义错误处理也就是本地错误处理，就是处理指定的通道。我们需要定义一个处理异常的方法，并在此方法上添加`@ServiceActivator`批注。批注具有`inputChannel`属性，该属性以`{destination}.{group}.errors`的形式指定要处理的通道。

代码如下：

@EnableBinding({ Sink.class})
    public class SinkReceiver { 
        private static final Logger LOGGER = LoggerFactory.getLogger(SinkReceiver.class);
    
        @StreamListener(Sink.INPUT)
        public void receive(Object payload) { 
            LOGGER.info("Received: {},{}", payload,new Date());
            throw new RuntimeException("error!");
        }
    
        /** * 消息消费失败的降级处理逻辑 * * @param message */
        @ServiceActivator(inputChannel = "greetings.stream-exception-handler.errors")
        public void error(Message<?> message) { 
            LOGGER.info("Message consumer failed, call fallback!");
        }
    }

运行结果如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70]

代码参考：[https://gitee.com/cckevincyh/cloud-stream-demo/tree/demo12/][https_gitee.com_cckevincyh_cloud-stream-demo_tree_demo12]

### 全局异常处理 ###

全局处理能够处理所有通道抛出的异常，所有通道抛出的异常都会生成一个ErrorMessage对象，即错误消息。错误消息放置在专用通道中，该专用通道是错误通道。因此，通过侦听errorChannel，可以处理全局异常。

代码如下：

@EnableBinding({ Sink.class})
    public class SinkReceiver { 
        private static final Logger LOGGER = LoggerFactory.getLogger(SinkReceiver.class);
    
        @StreamListener(Sink.INPUT)
        public void receive(Object payload) { 
            LOGGER.info("Received: {},{}", payload,new Date());
            throw new RuntimeException("error!");
        }
    
        /** * A Method of Handling Global Exceptions * * @param errorMessage Exception message object */
        @StreamListener("errorChannel")
        public void handleError(ErrorMessage errorMessage) { 
            LOGGER.error("exception occurred. errorMessage = {}", errorMessage);
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70 1]

代码参考：[https://gitee.com/cckevincyh/cloud-stream-demo/tree/demo14/][https_gitee.com_cckevincyh_cloud-stream-demo_tree_demo14]

### 死信队列 ###

配置如下：

spring:
     rabbitmq:
        host: 192.168.99.100
        port: 5672
        username: guest
        password: guest
     cloud:
       stream:
         bindings:
           input:
             destination: greetings
             group: stream-exception-handler
             consumer:
               max-attempts: 4
         rabbit:
           bindings:
             input:
               consumer:
                 auto-bind-dlq: true #开启DLQ（死信队列）

启动控制台如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70 2]

当出现异常之后就会把message丢入到死信队列

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70 3]

然后我们可以利用控制台的`Move messages`的功能把死信队列的message丢回到处理的队列中去。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70 4]

代码参考：[https://gitee.com/cckevincyh/cloud-stream-demo/tree/demo13/][https_gitee.com_cckevincyh_cloud-stream-demo_tree_demo13]

## 参考 ##

[Spring Cloud Stream 进阶配置——高可用（一）——失败重试][Spring Cloud Stream]

[Spring Cloud Stream消费失败后的处理策略（一）：自动重试][Spring Cloud Stream 1]

[Spring Cloud Stream消费失败后的处理策略（二）：自定义错误处理逻辑][Spring Cloud Stream 2]

[Spring Cloud Stream 进阶配置——高可用（二）——死信队列][Spring Cloud Stream 3]

[Spring Cloud Stream 重点与总结][Spring Cloud Stream 4]

[Spring Cloud Stream exception handling][]

[Spring Cloud Stream消费失败后的处理策略（三）：使用DLQ队列（RabbitMQ）][Spring Cloud Stream_DLQ_RabbitMQ]

[Spring Cloud Stream消费失败后的处理策略（四）：重新入队（RabbitMQ）][Spring Cloud Stream_RabbitMQ]

[https_gitee.com_cckevincyh_cloud-stream-demo_tree_demo11]: https://gitee.com/cckevincyh/cloud-stream-demo/tree/demo11/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70]: /images/20230208/e714f39a5dc64c2cadd15aa146b6b728.png
[https_gitee.com_cckevincyh_cloud-stream-demo_tree_demo12]: https://gitee.com/cckevincyh/cloud-stream-demo/tree/demo12/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70 1]: /images/20230208/e45bc60e954c47f88300d7566fef9a94.png
[https_gitee.com_cckevincyh_cloud-stream-demo_tree_demo14]: https://gitee.com/cckevincyh/cloud-stream-demo/tree/demo14/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70 2]: /images/20230208/98ee9760593b44138ed449a799bf6f1e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70 3]: /images/20230208/ac8bf98ad00548778e62434c4f467c16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Nja2V2aW5jeWg_size_16_color_FFFFFF_t_70 4]: /images/20230208/2a3f2d5f8cff46cda8a2ec743f0423f6.png
[https_gitee.com_cckevincyh_cloud-stream-demo_tree_demo13]: https://gitee.com/cckevincyh/cloud-stream-demo/tree/demo13/
[Spring Cloud Stream]: https://www.jianshu.com/p/d5acbf4572a1
[Spring Cloud Stream 1]: http://blog.didispace.com/spring-cloud-starter-finchley-7-2/
[Spring Cloud Stream 2]: http://blog.didispace.com/spring-cloud-starter-finchley-7-3/
[Spring Cloud Stream 3]: https://www.jianshu.com/p/d2ba4b988e97
[Spring Cloud Stream 4]: http://www.itmuch.com/spring-cloud/spring-cloud-stream-study/
[Spring Cloud Stream exception handling]: https://programmer.help/blogs/spring-cloud-stream-exception-handling.html
[Spring Cloud Stream_DLQ_RabbitMQ]: http://blog.didispace.com/spring-cloud-starter-finchley-7-4/
[Spring Cloud Stream_RabbitMQ]: http://blog.didispace.com/spring-cloud-starter-finchley-7-5/