一、实现思路
二、准备,框架搭建
/** * 消息发送历史 * * @author futao * @date 2020/3/31. */ @Getter @Setter @Builder @TableName("message") public class Message extends IdTimeEntity { @Tolerate public Message() { } /** * 消息承载的业务数据 */ @TableField("msg_data") private String msgData; /** * 交换机名称 */ @TableField("exchange_name") private String exchangeName; /** * 路由键 */ @TableField("routing_key") private String routingKey; /** * 消息状态 * * @see com.futao.springboot.learn.rabbitmq.doc.reliabledelivery.model.enums.MessageStatusEnum */ @TableField("status") private int status; /** * 重试次数 */ @TableField("retry_times") private int retryTimes; /** * 下一次重试时间 */ @TableField("next_retry_date_time") private LocalDateTime nextRetryDateTime; } 复制代码
/** * 消息状态枚举 * * @author futao * @date 2020/3/31. */ @Getter @AllArgsConstructor public enum MessageStatusEnum { /** * 1=发送中 */ SENDING(1, "发送中"), /** * 2=发送失败 */ SUCCESS(2, "发送成功"), /** * 3=发送失败 */ FAIL(3, "发送失败"); private int status; private String description; } 复制代码
spring: rabbitmq: host: localhost port: 5672 username: futao password: 123456789 virtual-host: reliable-delivery connection-timeout: 15000 # 发送确认 publisher-confirms: true # 路由失败回调 publisher-returns: true template: # 必须设置成true 消息路由失败通知监听者,而不是将消息丢弃 mandatory: true app: rabbitmq: retry: # 消息最大重试次数 max-retry-times: 5 # 每次重试时间间隔 retry-interval: 5s # 队列定义 queue: user: user-queue # 交换机定义 exchange: user: user-exchange 复制代码
三、编码
/** * RabbitMQ队列定义与绑定 * * @author futao * @date 2020/3/31. */ @Configuration public class Declare { @Bean public Queue userQueue(@Value("${app.rabbitmq.queue.user}") String userQueueName) { return QueueBuilder .durable(userQueueName) //.withArgument("x-max-length", 2) .build(); } @Bean public Exchange userExchange(@Value("${app.rabbitmq.exchange.user}") String userExchangeName) { return ExchangeBuilder .topicExchange(userExchangeName) .durable(true) .build(); } @Bean public Binding userBinding(Queue userQueue, Exchange userExchange) { return BindingBuilder .bind(userQueue) .to(userExchange) .with("user.*") .noargs(); } } 复制代码
confirmCallback()
消息投递回调方法与 returnCallback()
消息路由失败回调方法 /** * Bean增强 * 【严重警告】: 不可在该类中注入Bean,被注入的Bean不会被BeanPostProcessor增强,造成误伤。 * 必须通过容器来获取需要注入的Bean * * @author futao * @date 2020/3/20. */ @Slf4j @Component public class BeanEnhance implements BeanPostProcessor { // @Resource // private MessageMapper messageMapper; /** * 消息的最大重试次数 */ @Value("${app.rabbitmq.retry.max-retry-times}") private int maxRetryTimes; /** * 每次重试时间间隔 */ @Value("${app.rabbitmq.retry.retry-interval}") private Duration retryInterval; // @Autowired // private RabbitTemplate rabbitTemplate; // // @Autowired // private BeanEnhance enhance; @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { //增强RabbitTemplate if (RabbitTemplate.class.equals(bean.getClass())) { //消息投递成功与否的监听,可以用来保证消息100%投递到rabbitMQ。(如果某条消息(通过id判定)在一定时间内未收到该回调,则重发该消息) //需要设置 publisher-confirms: true ((RabbitTemplate) bean).setConfirmCallback((correlationData, ack, cause) -> { String correlationDataId = correlationData.getId(); if (ack) { //ACK log.debug("消息[{}]投递成功,将DB中的消息状态设置为投递成功", correlationDataId); ApplicationContextHolder.getBean(MessageMapper.class).update(null, Wrappers.<Message>lambdaUpdate() .set(Message::getStatus, MessageStatusEnum.SUCCESS.getStatus()) .eq(Message::getId, correlationDataId) ); } else { log.debug("消息[{}]投递失败,cause:{}", correlationDataId, cause); //NACK,消息重发 ApplicationContextHolder.getBean(BeanEnhance.class).reSend(correlationDataId); } }); //消息路由失败的回调--需要设置 publisher-returns: true 并且 template: mandatory: true 否则rabbit将丢弃该条消息 ((RabbitTemplate) bean).setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> { log.warn("消息路由失败回调...做一些补偿或者记录............................................."); log.warn("message{}", message); log.warn("replyCode{}", replyCode); log.warn("replyText{}", replyText); log.warn("exchange{}", exchange); log.warn("routingKey{}", routingKey); }); } return bean; } /** * NACK时进行消息重发 * * @param correlationDataId */ @Transactional(rollbackFor = Exception.class) public void reSend(String correlationDataId) { Message message = ApplicationContextHolder.getBean(MessageMapper.class).selectById(correlationDataId); if (message.getRetryTimes() < maxRetryTimes) { //进行重试 ApplicationContextHolder.getBean(RabbitTemplate.class).convertAndSend(message.getExchangeName(), message.getRoutingKey(), message.getMsgData(), new CorrelationData(correlationDataId)); //更新DB消息状态 ApplicationContextHolder.getBean(MessageMapper.class).update(null, Wrappers.<Message>lambdaUpdate() .set(Message::getStatus, MessageStatusEnum.SENDING.getStatus()) .set(Message::getNextRetryDateTime, LocalDateTime.now(ZoneOffset.ofHours(8)).plus(retryInterval)) .set(Message::getRetryTimes, message.getRetryTimes() + 1) .eq(Message::getId, correlationDataId) ); } } } 复制代码
/** * @author futao * @date 2020/3/31. */ @Component public class Sender { @Value("${app.rabbitmq.retry.retry-interval}") private Duration retryInterval; @Autowired private RabbitTemplate rabbitTemplate; @Resource private MessageMapper messageMapper; @Value("${app.rabbitmq.exchange.user}") private String userExchangeName; public void send(User user) { Message message = Message.builder() .msgData(JSON.toJSONString(user)) .exchangeName(userExchangeName) .routingKey("user.abc") .status(MessageStatusEnum.SENDING.getStatus()) //下次重试时间 .nextRetryDateTime(LocalDateTime.now(ZoneOffset.ofHours(8)).plus(retryInterval)) .retryTimes(0) .build(); //消息落库 messageMapper.insert( message ); CorrelationData correlationData = new CorrelationData(message.getId()); //消息投递到MQ rabbitTemplate.convertAndSend(userExchangeName, "user.abc", JSON.toJSONString(user), correlationData); } } 复制代码
xxl: job: switch: ON admin: ### 调度中心部署跟地址 [选填]:如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调";为空则关闭自动注册; addresses: http://127.0.0.1:9090/xxl-job-admin executor: ### 执行器AppName [选填]:执行器心跳注册分组依据;为空则关闭自动注册 appname: xxl-job-executor-rabbitmq ### 执行器IP [选填]:默认为空表示自动获取IP,多网卡时可手动设置指定IP,该IP不会绑定Host仅作为通讯实用;地址信息用于 "执行器注册" 和 "调度中心请求并触发任务"; # ip: ### 执行器端口号 [选填]:小于等于0则自动获取;默认端口为9999,单机部署多个执行器时,注意要配置不同执行器端口; port: 9999 ### 执行器运行日志文件存储磁盘路径 [选填] :需要对该路径拥有读写权限;为空则使用默认路径; logpath: data/applogs/xxl-job/jobhandler ### 执行器日志文件保存天数 [选填] : 过期日志自动清理, 限制值大于等于3时生效; 否则, 如-1, 关闭自动清理功能; logretentiondays: 30 ### 执行器通讯TOKEN [选填]:非空时启用; accessToken: 复制代码
/** * XXL-JOB配置 * * @author futao * @date 2020/4/1. */ @Setter @Getter @Slf4j @Configuration @ConfigurationProperties(prefix = "xxl.job") public class XxlJobConfig { private final Admin admin = new Admin(); private final Executor executor = new Executor(); @Bean public XxlJobSpringExecutor xxlJobExecutor(XxlJobConfig xxlJobConfig) { log.info(">>>>>>>>>>> xxl-job config init."); XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor(); xxlJobSpringExecutor.setAdminAddresses(xxlJobConfig.getAdmin().getAddresses()); xxlJobSpringExecutor.setAppName(xxlJobConfig.getExecutor().getAppName()); xxlJobSpringExecutor.setIp(xxlJobConfig.getExecutor().getIp()); xxlJobSpringExecutor.setPort(xxlJobConfig.getExecutor().getPort()); xxlJobSpringExecutor.setLogPath(xxlJobConfig.getExecutor().getLogPath()); xxlJobSpringExecutor.setLogRetentionDays(xxlJobConfig.getExecutor().getLogRetentionDays()); return xxlJobSpringExecutor; } @Getter @Setter public static class Admin { private String addresses; } @Getter @Setter public static class Executor { private String appName; private String ip; private int port; private String logPath; private int logRetentionDays; } } 复制代码
/** * 扫描数据库中需要重新投递的消息并重新投递 * * @author futao * @date 2020/4/1. */ @Slf4j @Component public class MessageReSendJob extends IJobHandler { @Autowired private RabbitTemplate rabbitTemplate; @Resource private MessageMapper messageMapper; @Autowired private MessageReSendJob messageReSendJob; /** * 最大重试次数 */ @Value("${app.rabbitmq.retry.max-retry-times}") private int retryTimes; /** * 重试时间间隔 */ @Value("${app.rabbitmq.retry.retry-interval}") private Duration retryInterval; /** * 批量从数据库中读取的消息 */ private static final int PAGE_SIZE = 100; @XxlJob(value = "MessageReSendJob", init = "jobHandlerInit", destroy = "jobHandlerDestroy") @Override public ReturnT<String> execute(String s) throws Exception { long startTime = System.currentTimeMillis(); log.info("开始扫描需要进行重试投递的消息"); XxlJobLogger.log("开始扫描需要进行重试投递的消息"); service(1); log.info("扫描需要进行重试投递的消息任务结束,耗时[{}]ms", System.currentTimeMillis() - startTime); XxlJobLogger.log("扫描需要进行重试投递的消息任务结束,耗时[{}]ms", System.currentTimeMillis() - startTime); return ReturnT.SUCCESS; } public void service(int pageNum) { IPage<Message> messageIPage = messageMapper.selectPage(new Page<>(pageNum, PAGE_SIZE), Wrappers.<Message>lambdaQuery() //发送中的消息 .eq(Message::getStatus, MessageStatusEnum.SENDING.getStatus()) //已到达下次发送时间 .le(Message::getNextRetryDateTime, LocalDateTime.now(ZoneOffset.ofHours(8))) ); List<Message> messages = messageIPage.getRecords(); messages.forEach(message -> { if (retryTimes <= message.getRetryTimes()) { //已达到最大投递次数,将消息设置为投递失败 messageMapper.update(null, Wrappers.<Message>lambdaUpdate().set(Message::getStatus, MessageStatusEnum.FAIL.getStatus()).eq(Message::getId, message.getId())); } else { messageReSendJob.reSend(message); } }); if (PAGE_SIZE == messages.size()) { service(++pageNum); } } /** * 重新投递消息 * * @param message */ public void reSend(Message message) { messageMapper.update(null, Wrappers.<Message>lambdaUpdate() .set(Message::getRetryTimes, message.getRetryTimes() + 1) .set(Message::getNextRetryDateTime, LocalDateTime.now(ZoneOffset.ofHours(8)).plus(retryInterval)) .eq(Message::getId, message.getId()) ); try { //再次投递 rabbitTemplate.convertAndSend(message.getExchangeName(), message.getRoutingKey(), message.getMsgData(), new CorrelationData(message.getId())); } catch (Exception e) { log.error("消息[{}]投递失败", JSON.toJSONString(message)); } } public void jobHandlerInit() { log.info("before job execute..."); XxlJobLogger.log("before job handler init..."); } public void jobHandlerDestroy() { log.info("after job execute..."); XxlJobLogger.log("after job execute..."); } } 复制代码
四、测试
/** * @author futao * @date 2020/4/1. */ @RequestMapping("/user") @RestController public class UserController { @Autowired private Sender sender; @RequestMapping("/send") public void send() { sender.send(User .builder() .userName("天文") .birthday(LocalDate.of(1995, 1, 31)) .address("浙江杭州") .build()); } } 复制代码
消息落库,状态为1=发送中
回调
消息设置成投递成功
启动生产者服务后停止MQ 发送消息
因为收不到该条消息的ACK。所以一直处于发送中。开启任务调度再次进行投递(投递次数+1,且更新下次投递时间)
当投递次数达到最大投递次数,下一次,将消息设置成投递失败
调度日志