本篇文章将介绍另一种高可用的服务架构,以便实现邮件 100% 被投递成功。类似的短信推送等服务,实现逻辑也大体类似。
图片
本文内容主要围绕这个流程图展开,利用 RabbitMQ 消息队列来实现邮件 100% 被投递,内容涵盖了 RabbitMQ 很多知识点,如:
生产者和消费者模型
消息发送机制
消费确认机制
消息的重新投递
消息消费失败的处理方案
1.准备一台电脑,并安装 RabbitMQ 服务
2.开放 QQ 邮箱或者其它邮箱授权码,用于发送邮件
3.创建邮件发送项目并编写代码
4.发送邮件测试
5.消息消费失败的处理介绍
安装 RabbitMQ 服务,这一步比较简单,可以访问下面的官方地址,下载软件包并依次按照步骤进行安装即可。
https://rabbitmq.org.cn/docs/download
安装成功之后,登陆 RabbitMQ 控制台,可以看到类似于如下界面。
图片
点击“Exchanges”菜单,进入“交换器”管理界面。
图片
进入之后,点击最下方“Add a new exchange”按钮,创建一个类型为topic,名称叫mail.exchange的交换器,并提交。
图片
接着,点击“Queues”菜单,进入消息队列管理界面。
图片
同样的,点击最下方“Add a new queue”按钮,创建一个名称叫mq.mail.ack的消息队列,并提交。
图片
保存之后,在列表中可以看到刚刚创建的消息队列,然后点击进入详情。
图片
在详情中,将当前消息队列与上文创建的交换器进行绑定,便于后续通过交换器来发送消息到队列,操作如下。
图片
对于topic类型的交换器,通常不直接与消息队列进行交互,而是通过一个路由键,将消息路由到目标消息队列,这样设计的目的是让消息投递更加灵活。路由键,可以简单理解为类似于路由器,对数据进行路由分发处理。
为了实现邮件自动发送功能,我们还需要准备一个邮箱发送服务器,这一步在之前的文章中已经详细的介绍过,在此,我们简单的再介绍一下。
以 QQ 邮箱为例,登陆进去之后,在设置里面开启 POP3/SMTP 服务,并获取授权码记录下来。
图片
图片
该授权码,就是下文配置文件中spring.mail.password需要的密码!
在 IDEA 下创建一个名称为smail的 Spring Boot 项目,pom文件中加入amqp和mail相关依赖包,示例如下:
<!--mail 支持--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-mail</artifactId></dependency><!--amqp 支持--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>
在application.properties中添加 rabbitmq、邮箱相关配置,示例如下:
# 配置邮件发送主机地址spring.mail.host=smtp.exmail.qq.com# 配置邮件发送服务端口号spring.mail.port=465# 配置邮件发送服务协议spring.mail.protocol=smtp# 配置邮件发送者用户名或者账户spring.mail.username=xxxx# 配置邮件发送者密码或者授权码spring.mail.password=xxxx# 配置邮件默认编码spring.mail.default-encoding=UTF-8# 配置smtp相关属性spring.mail.properties.mail.smtp.auth=truespring.mail.properties.mail.smtp.ssl.enable=truespring.mail.properties.mail.smtp.ssl.required=true#rabbitmq配置spring.rabbitmq.host=127.0.0.1spring.rabbitmq.port=5672spring.rabbitmq.virtual-host=/spring.rabbitmq.username=test spring.rabbitmq.password=test# 开启confirms回调 P -> Exchangespring.rabbitmq.publisher-cnotallow=true# 开启returnedMessage回调 Exchange -> Queuespring.rabbitmq.publisher-returns=true# 设置手动确认(ack) Queue -> Cspring.rabbitmq.listener.simple.acknowledge-mode=manual spring.rabbitmq.listener.simple.prefetch=100
其中,spring.mail.username和spring.mail.password指的就是上文中创建的邮箱账号和授权码,将其配置进去即可。
编写一个 RabbitMQ 配置类,用于监听消息的发送情况,示例如下。
@Configurationpublicclass RabbitConfig { private static final Logger LOGGER=LoggerFactory.getLogger(RabbitConfig.class);@Autowiredprivate CachingConnectionFactory connectionFactory;@BeanpublicRabbitTemplate rabbitTemplate(){ RabbitTemplate rabbitTemplate=new RabbitTemplate(connectionFactory);// 设置消息转换器为json格式rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());// 消息是否成功发送到ExchangerabbitTemplate.setConfirmCallback((correlationData,ack,cause)->{if(ack){ LOGGER.info("消息发送到Exchange成功,{}",correlationData);}else{ LOGGER.error("消息发送到Exchange失败, {}, cause: {}",correlationData,cause);} });// 触发setReturnCallback回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调rabbitTemplate.setMandatory(true);// 消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有消息从Exchange路由到Queue失败才会回调这个方法rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey)->{ LOGGER.error("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}",exchange,routingKey,replyCode,replyText,message);});returnrabbitTemplate;} }
在 Spring Boot 中,我们可以利用RabbitTemplate工具,将数据通过交换器发送到目标消息队列,示例如下。
@Servicepublicclass ProduceService {@Autowiredprivate RabbitTemplate rabbitTemplate;/** * 发送消息 * @param mail * @return */publicbooleansendByAck(Mail mail){// 创建uuidString msgId=UUID.randomUUID().toString().replaceAll("-","");mail.setMsgId(msgId);// 发送消息到mq服务器中(附带消息ID)CorrelationData correlationData=new CorrelationData(msgId);rabbitTemplate.convertAndSend("mail.exchange","route.mail.ack",MessageHelper.objToMsg(mail),correlationData);returntrue;} }
在 Spring Boot 中,我们可以利用@RabbitListener注解,监听指定的消息队列,如果队列中有消息会第一时间收到回调,示例如下。
@Componentpublicclass ConsumerService { private static final Logger LOGGER=LoggerFactory.getLogger(ConsumerService.class);@Autowiredprivate SendMailService sendMailService;/** * 监听消息队列,手动确认模式,必须手动调用ack或者nack方法 * 配置参数:spring.rabbitmq.listener.simple.acknowledge-mode=manual * @param message * @param channel * @throws IOException */@RabbitListener(queues={"mq.mail.ack"})publicvoid consumeFromAck(Message message,Channel channel)throws IOException { LOGGER.info("收到消息:{}",message.toString());//将消息转化为对象Mail mail=MessageHelper.msgToObj(message,Mail.class);// 手动确认模式long tag=message.getMessageProperties().getDeliveryTag();booleansuccess=sendMailService.send(mail);if(success){// 消费成功,消息会被删除channel.basicAck(tag,false);}else{// 消费失败,重新返回队列channel.basicNack(tag,false,true);} } }
正如之前的文章中所介绍的,在 Spring Boot 中,我们可以利用JavaMailSender工具来实现邮件的自动推送,示例如下。
@Servicepublicclass SendMailService { private static final Logger LOGGER=LoggerFactory.getLogger(SendMailService.class);@Value("${spring.mail.username}")private Stringfrom;@Autowiredprivate JavaMailSender mailSender;/** * 发送简单邮件 * * @param mail */publicbooleansend(Mail mail){ Stringto=mail.getTo();// 目标邮箱String title=mail.getTitle();// 邮件标题String content=mail.getContent();// 邮件正文SimpleMailMessage message=new SimpleMailMessage();message.setFrom(from);message.setTo(to);message.setSubject(title);message.setText(content);try { mailSender.send(message);LOGGER.info("邮件发送成功");returntrue;} catch(MailException e){ LOGGER.error("邮件发送失败, to: {}, title: {}",to,title,e);returnfalse;} } }
接着,编写一个 controller 接口,将邮件发送服务暴露出去,示例如下:
@RestControllerpublicclass MailController {@Autowiredprivate ProduceService produceService;@PostMapping("send")publicString sendMail(Mail mail){booleanresult=produceService.sendByAck(mail);returnresult ?"success":"fail";} }
最后,启动 SpringBoot 服务,用 postman 来测试一下。
图片
查看控制台信息。
图片
查询接受者邮件信息。
图片
可以清楚的看到,邮件发送成功!
当大批量的发送邮件,也不用担心,因为整个邮件的发送都是异步的,不会阻塞主流程的运行。
虽然以上的方案非常可靠,可以保证发出的消息 100% 被消费,但是其实也有弊端。
试想一下,按照上面的处理逻辑,假设其中有一条消息,因为某种原因一直发送失败,会出现什么样的情况?
此时,这条消息会重新返回队列,然后一直重试,会导致其它的消息可能会无法被消费。
针对这种情况,最简单粗暴的办法就是,当重试失败之后将消息丢弃,不会阻碍其它的消息被正常处理,不过会丢失数据。
那么如何正确的处理消息消费失败的问题呢?
可以借助数据库来记录消费失败的数据,针对系统无法成功处理的消息,人工进行干预。
实践过程如下!
首先,在数据库中创建一张消息日志表,用于跟踪消息数据的状态,示例如下:
CREATETABLE`msg_log`(`msg_id`varchar(255)NOTNULLDEFAULT''COMMENT'消息唯一标识',`exchange`varchar(100)NOTNULLDEFAULT''COMMENT'交换机',`route_key`varchar(100)NOTNULLDEFAULT''COMMENT'路由键',`queue_name`varchar(100)NOTNULLDEFAULT''COMMENT'队列名称',`msg`textCOMMENT'消息体, json格式化',`result`varchar(255)DEFAULTNULLCOMMENT'处理结果',`status`int(11)NOTNULLDEFAULT'0'COMMENT'状态,0:等待消费,1:消费成功,2:消费失败,9:重试失败',`try_count`int(11)NOTNULLDEFAULT'0'COMMENT'重试次数',`next_try_time`datetimeDEFAULTNULLCOMMENT'下一次重试时间',`create_time`datetimeDEFAULTNULLCOMMENT'创建时间',`update_time`datetimeDEFAULTNULLCOMMENT'更新时间',PRIMARYKEY(`msg_id`))ENGINE=InnoDBDEFAULTCHARSET=utf8mb4COMMENT='mq消息日志';
在生产者服务类中,先将消息数据写入数据库,再向 rabbitMQ 服务中发消息,示例如下:
@Servicepublicclass ProduceService {@Autowiredprivate RabbitTemplate rabbitTemplate;@Autowiredprivate MsgLogService msgLogService;/** * 发送消息 * @param mail * @return */publicbooleansendByAuto(Mail mail){ String msgId=UUID.randomUUID().toString().replaceAll("-","");mail.setMsgId(msgId);// 1.存储要消费的数据msgLogService.save("mail.exchange","route.mail.auto","mq.mail.auto",msgId,mail);// 2.发送消息到mq服务器中(附带消息ID)CorrelationData correlationData=new CorrelationData(msgId);rabbitTemplate.convertAndSend("mail.exchange","route.mail.auto",MessageHelper.objToMsg(mail),correlationData);returntrue;} }
在消费者服务类中,收到消息之后,不管处理成功还是失败,都只会修改数据库中的消息状态,并且消息处理失败时,不再重新返回队列。
@Componentpublicclass ConsumerService { private static final Logger LOGGER=LoggerFactory.getLogger(ConsumerService.class);@Autowiredprivate SendMailService sendMailService;@Autowiredprivate MsgLogService msgLogService;/** * 监听消息队列,自动确认模式,无需调用ack或者nack方法,当程序执行时才删除消息 * 配置参数:spring.rabbitmq.listener.simple.acknowledge-mode=auto * @param message */@RabbitListener(queues={"mq.mail.auto"})publicvoid consumeFromAuto(Message message){ LOGGER.info("收到消息:{}",message.toString());// 获取消息IDMail mail=MessageHelper.msgToObj(message,Mail.class);// 消息幂等性处理,如果已经处理成功,无需重复消费MsgLog queryObj=msgLogService.selectByMsgId(mail.getMsgId());if(Objects.nonNull(queryObj)&&Constant.SUCCESS.equals(queryObj.getStatus())){return;}// 发送邮件booleansuccess=sendMailService.send(mail);if(success){ msgLogService.updateStatus(mail.getMsgId(),Constant.SUCCESS,"邮件发送成功");}else{ msgLogService.updateStatus(mail.getMsgId(),Constant.FAIL,"邮件发送失败");} } }
因为此处采用自动确认模式,因此还需要修改application.properties中的配置参数,内容如下:
# 设置自动确认(默认此模式)spring.rabbitmq.listener.simple.acknowledge-mode=auto
当消息消费失败时,会自动记录到数据库。
实际上,不可能每条数据都需要我们进行干预,有的可能重试一次就好了,因此可以编写一个定时任务,将消费失败的数据筛选出来,重新放入到消息队列中,只有当消费次数达到设置的最大值,此时进入人工干预阶段,可以节省不少的工作。
示例如下:
@Componentpublicclass ScheduledTask { private static final Logger LOGGER=LoggerFactory.getLogger(ScheduledTask.class);/** * 最大投递次数 */private static finalintMAX_TRY_COUNT=3;@Autowiredprivate MsgLogService msgLogService;@Autowiredprivate RabbitTemplate rabbitTemplate;/** * 每30s拉取消费失败的消息, 重新投递 */@Scheduled(cron="0/30 * * * * ?")publicvoid retry(){ LOGGER.info("开始执行重新投递消费失败的消息!");// 查询需要重新投递的消息List<MsgLog>msgLogs=msgLogService.selectFailMsg();for(MsgLog msgLog : msgLogs){if(msgLog.getTryCount()>=MAX_TRY_COUNT){ msgLogService.updateStatus(msgLog.getMsgId(),Constant.RETRY_FAIL,msgLog.getResult());LOGGER.info("超过最大重试次数, msgId: {}",msgLog.getMsgId());break;}// 重新投递消息CorrelationData correlationData=new CorrelationData(msgLog.getMsgId());rabbitTemplate.convertAndSend("",msgLog.getQueueName(),MessageHelper.objToMsg(msgLog.getMsg()),correlationData);// 更新下次重试时间msgLogService.updateNextTryTime(msgLog.getMsgId(),msgLog.getTryCount());} } }
最后别忘了,在Application类上添加@EnableScheduling,以便让定时调度生效,示例如下:
@EnableScheduling @SpringBootApplication public class Application { public static void main(String[] args) { SpringApplication.run(Application.class,args); } }
利用定时任务,对投递失败的消息进行补偿投递,基本可以保证消息 100% 消费成功!
本文主要以实现邮件自动推送这个业务场景为例,通过 Springboot 整合 rabbitMQ 技术来实现高可用的效果。
当然,解决这个业务需求的技术方案还有很多,例如 Springboot 整合 rocketMQ 也可以实现这个效果,不管怎么变,底层的实现思路基本都一样。
希望本篇的知识总结,对大家有所帮助。
最后,代码都经过自测,想要获取项目源码的同学,可以点击如下地址获取。
示例代码地址:
https://gitee.com/pzblogs/spring-boot-example-demo