SpringCloud之Stream-8.Stream本地重试功

如果要进行重试的实验，那么就必须要使得程序报错

image.png

1.创建一个新的topic

public interface ErrorTopic {

    String INPUT = "error-consumer";

    String OUTPUT = "error-producer";

    @Input(INPUT)
    SubscribableChannel input();

    @Output(OUTPUT)
    MessageChannel output();
}

2.controller

 //异常重试（单机版）
    @PostMapping("sendErrorMessage")
    public void sendErrorMessage(@RequestParam(value="body")String body){
        errorTopicProducer.output().send(MessageBuilder.withPayload(body).build());
    }

3.监听器

private AtomicInteger count = new AtomicInteger(1);

//异常重试（单机版）
    @StreamListener(ErrorTopic.INPUT)
    public void consumeErrorMessage(Object payload){
        log.info("this is normal msg!");

        if(count.incrementAndGet()%3==0){
            log.info("fine,thank you");
            count.set(0);
        }else{
            log.info("what's your problem?");
            throw new RuntimeException("i am not ok");
        }
        log.info("Error message consumed successfully,payload={}",payload);
    }

4.配置

#异常消息（单机重试版本）
spring.cloud.stream.bindings.error-consumer.destination=error-out-topic
spring.cloud.stream.bindings.error-producer.destination=error-out-topic
#重试次数（本机重试） 次数为1相当于不重试
spring.cloud.stream.bindings.error-consumer.consumer.max-attempts=2

5.发送请求，测试

image.png image.png

可以看到当我们重试次数是两次的时候，那么第一次count应该是2，会进入到相应的逻辑去，但是并不会抛异常，然后会进行第二次重试；

如果重试次数是3的话，那么最后一次是会抛出异常的。