- application.properties中配置开启手动签收
spring.rabbitmq.listener.direct.acknowledge-mode=manual
spring.rabbitmq.listener.simple.acknowledge-mode=manual
- 定义一个重试器
@Slf4j
@Configuration
public class RabbitMQRetryConfing {@Bean("customRetry")public RetryTemplate retryTemplate() {RetryTemplate retryTemplate = new RetryTemplate();SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();retryPolicy.setMaxAttempts(3); // 设置重试次数retryTemplate.setRetryPolicy(retryPolicy);FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();backOffPolicy.setBackOffPeriod(2000); // 重试间隔，单位：毫秒retryTemplate.setBackOffPolicy(backOffPolicy);// 添加 RetryListener 以观察重试过程retryTemplate.registerListener(new RetryListener() {@Overridepublic <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {return true;}@Overridepublic <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {}@Overridepublic <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {//春姐，这里可以灵活配置重试失败后的回调，例如发送告警、更新数据库状态等if (context.getRetryCount() >= retryPolicy.getMaxAttempts()) {int retryCount = context.getRetryCount();// 当重试次数耗尽时进行处理log.info("已达到最大重试次数{}次，丢弃本次任务",retryCount);}}});return retryTemplate;}
}
这个重试器可以配置最大重试次数、重试之间间隔次数等策略配置，调用重试器的execute方法可以进行队列消费，如果在执行一次任务期间发生了异常，则会根据配置的重试次数以及间隔时间自动触发下一次重试，每一次重试都是在同一个线程中执行完成的，并且RetryTemplate会为每一次重试失败进行回调，提供了诸如 onOpen、onClose、onError等回调时机。
- RabbitMQ监听器回调方法
@Autowired@Qualifier("customRetry")private RetryTemplate retryTemplate;//queues消费的队列 ackMode确认模式 MANUAL 手动确认@RabbitListener(queues = "q",ackMode = "MANUAL")@Overridepublic void onMessage(Message message, Channel channel) {try {retryTemplate.execute(new RetryCallback<Object, Throwable>() {@Overridepublic Object doWithRetry(RetryContext retryContext) throws Throwable {// 消息的唯一标识idlong deliveryTag = message.getMessageProperties().getDeliveryTag();log.info("接收到mq信息" + new String(message.getBody()));try{//todo 开始业务处理String msg = new String(message.getBody());Integer articleId = Integer.parseInt(msg);newsService.pullNews(articleId);// 手动签收的第一个参数为消息的唯一标识id、第二个参数表示是否批量签收channel.basicAck(deliveryTag,false);log.info("消费mq消息成功，articleId为:{}",articleId);return null;}catch (Throwable e){log.error(String.format("失败，异常信息为：%s",new String(message.getBody()),e.getMessage()));//重新抛出异常 触发重试机制throw e;}finally {//重试次数达到限制log.error(String.format("失败",new String(message.getBody())));//不重新入队，发送到死信队列// 手动拒绝签收的第一个参数为消息id、// 第二个参数表示是否批量签收// 第三个参数消息是否重回队列channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);}}});} catch (Throwable throwable) {log.info("执行回调重试上下文出现异常");}}
这样，当耗尽完重试次数之后就会被回调到onError方法中执行自定义的异常逻辑处理。
