项目整体结构
本文涵盖了关于RabbitMQ很多方面的知识点, 如:
- 消息发送确认机制
- 消费确认机制
- 消息的重新投递
- 消费幂等性, 等等
这些都是围绕上面那张整体流程图展开的, 所以有必要先贴出来, 见图知意
实现思路
- 简略介绍 QQ邮箱授权码的获取
- 编写发送邮件工具类
- 编写 RabbitMQ 配置文件
- 生产者发起调用
- 消费者发送邮件
- 定时任务定时拉取投递失败的消息, 重新投递
各种异常情况的测试验证
表结构
CREATE TABLE `msg_log` (
`msg_id` varchar(255) NOT NULL DEFAULT '' COMMENT '消息唯一标识',
`msg` text COMMENT '消息体, json格式化',
`exchange` varchar(255) NOT NULL DEFAULT '' COMMENT '交换机',
`routing_key` varchar(255) NOT NULL DEFAULT '' COMMENT '路由键',
`status` int(11) NOT NULL DEFAULT '0' COMMENT '状态: 0投递中 1投递成功 2投递失败 3已消费',
`try_count` int(11) NOT NULL DEFAULT '0' COMMENT '重试次数',
`next_try_time` datetime DEFAULT NULL COMMENT '下一次重试时间',
`create_time` datetime DEFAULT NULL COMMENT '创建时间',
`update_time` datetime DEFAULT NULL COMMENT '更新时间',
PRIMARY KEY (`msg_id`),
UNIQUE KEY `unq_msg_id` (`msg_id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='消息投递日志';
项目介绍
springboot版本 2.6.1, 旧版本可能有些配置属性不能使用, 需要以代码形式进行配置
- RabbitMQ版本 3.7.15
- MailUtil: 发送邮件工具类
- RabbitConfig: rabbitmq 相关配置
- TestServiceImpl: 生产者, 发送消息
- MailConsumer: 消费者, 消费消息, 发送邮件
- ResendMsg: 定时任务, 重新投递发送失败的消息
项目实现
maven 依赖
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.4</version>
</dependency>
<!-- springboot-aop -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<!-- guava -->
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>26.0-jre</version>
</dependency>
<!--joda time-->
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
<version>2.10</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.9.2</version>
<!--移除1.5.20-->
<exclusions>
<exclusion>
<groupId>io.swagger</groupId>
<artifactId>swagger-models</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.7.22</version>
</dependency>
<!--引入1.5.21-->
<dependency>
<groupId>io.swagger</groupId>
<artifactId>swagger-models</artifactId>
<version>1.5.20</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-mail</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-collections4</artifactId>
<version>4.1</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<version>1.1.20</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--mybatis-->
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>2.0.0</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
application.yml 文件
spring:
# mvc这部分解决swagger3在新版本Springboot中无法展示的问题
mvc:
pathmatch:
matching-strategy: ANT_PATH_MATCHER
mail:
username: [email protected]
password: ayiojlyasreljfdb
from: [email protected]
host: smtp.qq.com
default-encoding: UTF-8
properties:
mail:
smtp:
auth: true
starttls:
enable: true
required: true
ssl:
enable: true
# mysql
datasource:
username: root
password: zgf123
url: jdbc:mysql://localhost:3306/test?serverTimezone=GMT%2B8&autoR&useUnicode=true&characterEncoding=utf-8&allowMultiQueries=true
type: com.alibaba.druid.pool.DruidDataSource
driver-class-name: com.mysql.cj.jdbc.Driver
# rabbitmq
rabbitmq:
host: 8.130.27.55
port: 5672
username: admin
password: 123
publisher-returns: true #returnedMessage Exchange -> Queue
listener:
simple:
acknowledge-mode: manual #表示开启手动ACK确认消息。
prefetch: 100 #表示每个消费者最大未确认(unack)的消息数量为100。
publisher-confirm-type: correlated #表相关确认模式,为每个消息设置唯一ID,回调时带上ID信息
mybatis:
mapper-locations: classpath:mapper/*.xml
server:
port: 8082
MailUtil 工具类
@Value("${spring.mail.from}")
private String from;
@Autowired
private JavaMailSender mailSender;
/**
* 发送简单邮件
*
* @param mail
*/
public boolean send(Mail mail) {
String to = 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);
log.info("邮件发送成功");
return true;
} catch (MailException e) {
log.error("邮件发送失败, to: {}, title: {}", to, title, e);
return false;
}
}
RabbitConfig 配置类
@Configuration
@Slf4j
public class RabbitConfig {
/**
* CachingConnectionFactory的主要功能有:
*
* 缓存RabbitMQ的连接,重复使用,避免频繁创建新连接。
* 支持CHANNEL的缓存,重复利用CHANNEL。
* 如果连接断开,能够自动重新连接。
* 提供设置连接缓存大小的方法。
* 支持在非事务模式和事务模式下使用。
*/
@Autowired
private CachingConnectionFactory connectionFactory;
@Autowired
private MsgLogService msgLogService;
@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
// 设置将对象转换为JSON字符串,所以发送消息时对象会被转为JSON。
rabbitTemplate.setMessageConverter(converter());
// 消息是否成功发送到Exchange
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
log.info("消息成功发送到Exchange");
String msgId = correlationData.getId();
// 发送成功,修改状态为已经投送
msgLogService.updateStatus(msgId, Constant.MsgLogStatus.DELIVER_SUCCESS);
} else {
log.info("消息发送到Exchange失败, {}, cause: {}", correlationData, cause);
String msgId = correlationData.getId();
msgLogService.updateStatus(msgId, Constant.MsgLogStatus.DELIVER_FAIL);
}
});
// 触发setReturnCallback回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调
rabbitTemplate.setMandatory(true);
// 消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有消息从Exchange路由到Queue失败才会回调这个方法
rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}", exchange, routingKey, replyCode, replyText, message);
});
return rabbitTemplate;
}
@Bean
public Jackson2JsonMessageConverter converter() {
return new Jackson2JsonMessageConverter();
}
// 登录日志
public static final String LOGIN_LOG_QUEUE_NAME = "login.log.queue";
public static final String LOGIN_LOG_EXCHANGE_NAME = "login.log.exchange";
public static final String LOGIN_LOG_ROUTING_KEY_NAME = "login.log.routing.key";
// 声明队列,登录日志队列
@Bean
public Queue logUserQueue() {
return new Queue(LOGIN_LOG_QUEUE_NAME, true);
}
// 声明登录日志交换机
@Bean
public DirectExchange logUserExchange() {
return new DirectExchange(LOGIN_LOG_EXCHANGE_NAME, true, false);
}
// 确定绑定关系
@Bean
public Binding logUserBinding() {
return BindingBuilder.bind(logUserQueue()).to(logUserExchange()).with(LOGIN_LOG_ROUTING_KEY_NAME);
}
// 发送邮件
public static final String MAIL_QUEUE_NAME = "mail.queue";
public static final String MAIL_EXCHANGE_NAME = "mail.exchange";
public static final String MAIL_ROUTING_KEY_NAME = "mail.routing.key";
// 定时发送邮件
public static final String MAIL_QUEUE_NAME_Q1 = "mail.queue.q1";
public static final String MAIL_EXCHANGE_NAME_E1 = "mail.exchange.e1";
public static final String MAIL_ROUTING_KEY_NAME_K1 = "mail.routing.key.k1";
// 死信延期
public static final String MAIL_QUEUE_NAME_DEL = "mail.queue.del";
public static final String MAIL_EXCHANGE_NAME_DEL = "mail.exchange.del";
public static final String MAIL_ROUTING_KEY_NAME_DEL = "mail.routing.key.del";
@Bean
public Queue mainQueueDel(){
return QueueBuilder.durable(MAIL_QUEUE_NAME_DEL).build();
}
@Bean
public DirectExchange mailExchangeDel(){
return new DirectExchange(MAIL_EXCHANGE_NAME_DEL,true,false);
}
@Bean
public Binding mailBindingDel() {
return BindingBuilder.bind(mainQueueDel()).to(mailExchangeDel()).with(MAIL_ROUTING_KEY_NAME_DEL);
}
@Bean
public Queue mailQueueQ1(){
Map<String,Object> arguments = new HashMap<>(3);
// 设置 死信交换机
arguments.put("x-dead-letter-exchange",MAIL_EXCHANGE_NAME_DEL);
// 设置死信routingkey
arguments.put("x-dead-letter-routing-key",MAIL_ROUTING_KEY_NAME_DEL);
return QueueBuilder.durable(MAIL_QUEUE_NAME_Q1).withArguments(arguments).build();
}
@Bean
public DirectExchange mailExchangeE1(){
return new DirectExchange(MAIL_EXCHANGE_NAME_E1,true,false);
}
@Bean
public Binding mailBindingE() {
return BindingBuilder.bind(mailQueueQ1()).to(mailExchangeE1()).with(MAIL_ROUTING_KEY_NAME_K1);
}
@Bean
public Queue mailQueue() {
return new Queue(MAIL_QUEUE_NAME, true);
}
@Bean
public DirectExchange mailExchange() {
return new DirectExchange(MAIL_EXCHANGE_NAME, true, false);
}
@Bean
public Binding mailBinding() {
return BindingBuilder.bind(mailQueue()).to(mailExchange()).with(MAIL_ROUTING_KEY_NAME);
}
}
TestServiceImpl生产消息
/**
* 发送消息
* @param mail 消息
* @return
*/
@Override
public ServerResponse send(Mail mail) {
// 生成一个唯一的消息id
String msgId = RandomUtil.UUID32();
mail.setMsgId(msgId);
// 构建消息
MsgLog msgLog = new MsgLog(msgId, mail, RabbitConfig.MAIL_EXCHANGE_NAME, RabbitConfig.MAIL_ROUTING_KEY_NAME);
msgLogMapper.insert(msgLog);// 消息入库
CorrelationData correlationData = new CorrelationData(msgId);
// 发送消息
rabbitTemplate.convertAndSend(RabbitConfig.MAIL_EXCHANGE_NAME, RabbitConfig.MAIL_ROUTING_KEY_NAME, MessageHelper.objToMsg(mail), correlationData);// 发送消息
return ServerResponse.success(ResponseCode.MAIL_SEND_SUCCESS.getMsg());
}
MailConsumer消费消息, 发送邮件
@Autowired
private MsgLogService msgLogService;
@Autowired
private MailUtil mailUtil;
/**
* 监听队列信息
* @param message 消息体
* @param channel 连接频道
* @throws IOException
*/
@RabbitListener(queues = RabbitConfig.MAIL_QUEUE_NAME)
public void consume(Message message, Channel channel) throws IOException {
// 将消息转为实体类
Mail mail = MessageHelper.msgToObj(message, Mail.class);
log.info("收到消息: {}", mail.toString());
String msgId = mail.getMsgId();
// 去数据库查询mq发送的消息是否存在
MsgLog msgLog = msgLogService.selectByMsgId(msgId);
// 如果不存在,或者状态标记为已消费,则不处理
if (null == msgLog || msgLog.getStatus().equals(Constant.MsgLogStatus.CONSUMED_SUCCESS)) {
log.info("重复消费, msgId: {}", msgId);
return;
}
// 获取消息唯一id
MessageProperties properties = message.getMessageProperties();
long tag = properties.getDeliveryTag();
// 发送邮件
boolean success = mailUtil.send(mail);
if (success) {
// 成功了修改状态,并手动应答
msgLogService.updateStatus(msgId, Constant.MsgLogStatus.CONSUMED_SUCCESS);
channel.basicAck(tag, false); // false 不批量确认
} else {
// 失败重新入队
channel.basicNack(tag, false, true);
}
}
说明: 其实就完成了 3 件事: 1. 保证消费幂等性, 2. 发送邮件, 3. 更新消息状态, 手动 ack
ResendMsg定时任务重新投递发送失败的消息
@Component
@Slf4j
public class ResendMsg {
@Autowired
private MsgLogService msgLogService;
@Autowired
private RabbitTemplate rabbitTemplate;
// 最大投递次数
private static final int MAX_TRY_COUNT = 3;
/**
* 每30s拉取投递失败的消息, 重新投递
*/
@Scheduled(cron = "0/30 * * * * ?")
public void resend() {
log.info("开始执行定时任务(重新投递消息)");
List<MsgLog> msgLogs = msgLogService.selectTimeoutMsg();
msgLogs.forEach(msgLog -> {
String msgId = msgLog.getMsgId();
if (msgLog.getTryCount() >= MAX_TRY_COUNT) {
// 查询投递失败的消息
msgLogService.updateStatus(msgId, Constant.MsgLogStatus.DELIVER_FAIL);
log.info("超过最大重试次数, 消息投递失败, msgId: {}", msgId);
} else {
msgLogService.updateTryCount(msgId, msgLog.getNextTryTime());// 投递次数+1
CorrelationData correlationData = new CorrelationData(msgId);
rabbitTemplate.convertAndSend(msgLog.getExchange(), msgLog.getRoutingKey(), MessageHelper.objToMsg(msgLog.getMsg()), correlationData);// 重新投递
log.info("第 " + (msgLog.getTryCount() + 1) + " 次重新投递消息");
}
});
log.info("定时任务执行结束(重新投递消息)");
}
}
说明: 每一条消息都和 exchange routingKey绑定, 所有消息重投共用这一个定时任务即可
基本测试
接口测试
数据库
状态为 3, 表明已消费, 消息重试次数为 0, 表明一次投递就成功了
参考:https://blog.csdn.net/androidstarjack/article/details/130073276
标签:return,NAME,rabbit,MAIL,mail,定时,public,邮件,String From: https://www.cnblogs.com/zgf123/p/17683946.html