重试策略
重试策略就是消费失败后隔一段时间我在消费，这种方案一般是硬卧环境因素导致的失败情况，或者网络问题导致的消费失败，重试消费可能解决上述问题。
配置参数
重试策略是通过配置参数实现的，参数前缀是spring.cloud.strema.bingings.<bindingName>.consumer. ，具体配置参数如下：
maxAttempts: 3 #对输入通道消息处理的最大重试次数，默认是3 次。
backOffMaxInterval: 10000 #重试消息处理的最大时间间隔
backOffInitialInterval: 1000 #重试消息处理的初始间隔时间。
通过配置上述参数就能使使重试策略生效。
代码实践
```
package com.dy.producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@EnableBinding(TestApplication.TestTopic.class)
@SpringBootApplication
public class TestApplication {
private static final Loggerlog= LoggerFactory.getLogger(TestApplication.class);
public static void main(String[] args) {
SpringApplication.run(TestApplication.class, args);
}
@RestController
static class TestController {
@Autowired
private TestTopictestTopic;
/**
* 消息生产接口
* @param message
* @return
*/
@GetMapping("/sendMessage")
public StringmessageWithMQ(@RequestParam String message) {
testTopic.output().send(MessageBuilder.withPayload(message).build());
return "ok";
}
}
/**
* 消息消费逻辑
*/
@Component
static class TestListener {
@StreamListener(TestTopic.INPUT)
public void receive(String payload) {
log.info("Received: " + payload);
throw new RuntimeException("Message consumer failed!");
}
}
interface TestTopic {
StringOUTPUT ="example-topic-output";
StringINPUT ="example-topic-input";
@Output(OUTPUT)
MessageChanneloutput();
@Input(INPUT)
SubscribableChannelinput();
}
}
配置参数
