RabbitMQ邮件发送

RabbitMQ流程
- 生产者:负责发送消息到Exchange
- Exchange:按照一定的策略,负责将消息存放到指定的队列
- 队列queue:负责保存消息
- 消费者:负责将队列中的消息提取
- binding:负责Exchange和队列的关联映射
RabbitMQ提供了4种不同的Exchange策略,分别是Direct,Fanout,Topic,Header
- Direct:DriectExchange是将消息队列绑定到一个DriectExchange上,通过routing-Key进行进行绑定,实现Exchange到Queue消息传递
- Fanout:Fanout的数据交换策略是把所有到达FanoutExchange的消息转发给与它绑定的Queue,routing key不起作用
- Topic:该策略比较灵活,Queue通过routing key绑定到TopicExchange上,当消息到达TopicExchange通过routingkey将消息路由到一个或多个Queue上
- Header:用的较少。。。
1.关于邮件的有两个实体类
- MailConstants:定义了很多关于邮件的常量
public class MailConstants {
public static final Integer DELIVERING = 0;//消息投递中
public static final Integer SUCCESS = 1;//消息投递成功
public static final Integer FAILURE = 2;//消息投递失败
public static final Integer MAX_TRY_COUNT = 3;//最大重试次数
public static final Integer MSG_TIMEOUT = 1;//消息超时时间
public static final String MAIL_QUEUE_NAME = "javaboy.mail.queue";
public static final String MAIL_EXCHANGE_NAME = "javaboy.mail.exchange";
public static final String MAIL_ROUTING_KEY_NAME = "javaboy.mail.routing.key";
}
- MailSendLog:定义了一个数据表,接着写了一个接口和xml配置文件访问数据库
public class MailSendLog {
private String msgId;
private Integer empId;
//0 消息投递中 1 投递成功 2投递失败
private Integer status;
private String routeKey;
private String exchange;
private Integer count;
private Date tryTime;
private Date createTime;
private Date updateTime;
}
public interface MailSendLogMapper {
Integer updateMailSendLogStatus(@Param("msgId") String msgId, @Param("status") Integer status);
Integer insert(MailSendLog mailSendLog);
List<MailSendLog> getMailSendLogsByStatus();
Integer updateCount(@Param("msgId") String msgId, @Param("date") Date date);
}
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper
PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="org.javaboy.vhr.mapper.MailSendLogMapper">
<update id="updateMailSendLogStatus">
update mail_send_log set status = #{status} where msgId=#{msgId};
</update>
<insert id="insert" parameterType="org.javaboy.vhr.model.MailSendLog">
insert into mail_send_log (msgId,empId,routeKey,exchange,tryTime,createTime) values (#{msgId},#{empId},#{routeKey},#{exchange},#{tryTime},#{createTime});
</insert>
<select id="getMailSendLogsByStatus" resultType="org.javaboy.vhr.model.MailSendLog">
select * from mail_send_log where status=0 and tryTime < sysdate()
</select>
<update id="updateCount">
update mail_send_log set count=count+1,updateTime=#{date} where msgId=#{msgId};
</update>
</mapper>
2.在业务层建立了MailSendLogService
@Service
public class MailSendLogService {
@Autowired
MailSendLogMapper mailSendLogMapper;
public Integer updateMailSendLogStatus(String msgId, Integer status) {
return mailSendLogMapper.updateMailSendLogStatus(msgId, status);
}
public Integer insert(MailSendLog mailSendLog) {
return mailSendLogMapper.insert(mailSendLog);
}
public List<MailSendLog> getMailSendLogsByStatus() {
return mailSendLogMapper.getMailSendLogsByStatus();
}
public Integer updateCount(String msgId, Date date) {
return mailSendLogMapper.updateCount(msgId,date);
}
}
3.对于MailSendLogService没有对应的Controller层,因此,在Service层建立了一个MailSendTask,这个类用于处理没发送成功
@Component
public class MailSendTask {
@Autowired
MailSendLogService mailSendLogService;
@Autowired
RabbitTemplate rabbitTemplate;
@Autowired
EmployeeService employeeService;
//实现定时:每10秒钟执行一次
@Scheduled(cron = "0/10 * * * * ?")
public void mailResendTask() {
List<MailSendLog> logs = mailSendLogService.getMailSendLogsByStatus();
if (logs == null || logs.size() == 0) {
return;
}
logs.forEach(mailSendLog->{
if (mailSendLog.getCount() >= 3) {
mailSendLogService.updateMailSendLogStatus(mailSendLog.getMsgId(), 2);//直接设置该条消息发送失败
}else{
mailSendLogService.updateCount(mailSendLog.getMsgId(), new Date());
Employee emp = employeeService.getEmployeeById(mailSendLog.getEmpId());
rabbitTemplate.convertAndSend(MailConstants.MAIL_EXCHANGE_NAME, MailConstants.MAIL_ROUTING_KEY_NAME, emp, new CorrelationData(mailSendLog.getMsgId()));
}
});
}
}
4.怎样利用RabbitMQ作为消息中间件实现注册消息的存储和转发
这个在employeeService这个类中的add方法中,在确认add成功之后,把employee中重要的信息放入RabbitMQ中的exchange中,并指定rounting-key。具体见下:
public Integer addEmp(Employee employee){
Date beginContract = employee.getBeginContract();
Date endContract = employee.getEndContract();
double month = (Double.parseDouble(yearFormat.format(endContract))- Double.parseDouble(yearFormat.format(beginContract)))* 12 + (Double.parseDouble(monthFormat.format(endContract)) - Double.parseDouble(monthFormat.format(beginContract)));
int result = employeeMapper.insertSelective(employee);
if (result == 1) {
Employee emp = employeeMapper.getEmployeeById(employee.getId());
//生成消息的唯一id
String msgId = UUID.randomUUID().toString();
MailSendLog mailSendLog = new MailSendLog();
mailSendLog.setMsgId(msgId);
mailSendLog.setCreateTime(new Date());
mailSendLog.setExchange(MailConstants.MAIL_EXCHANGE_NAME);
mailSendLog.setRouteKey(MailConstants.MAIL_ROUTING_KEY_NAME);
mailSendLog.setEmpId(emp.getId());
mailSendLog.setTryTime(new Date(System.currentTimeMillis() + 1000 * 60 * MailConstants.MSG_TIMEOUT));
mailSendLogService.insert(mailSendLog);
rabbitTemplate.convertAndSend(MailConstants.MAIL_EXCHANGE_NAME, MailConstants.MAIL_ROUTING_KEY_NAME, emp, new CorrelationData(msgId));
}
return result;
}
5.来看看RabbitConfig
@Configuration
public class RabbitConfig {
public final static Logger logger = LoggerFactory.getLogger(RabbitConfig.class);
@Autowired
CachingConnectionFactory cachingConnectionFactory;
@Autowired
MailSendLogService mailSendLogService;
@Bean
RabbitTemplate rabbitTemplate() {
RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
rabbitTemplate.setConfirmCallback((data, ack, cause) -> {
String msgId = data.getId();
if (ack) {
logger.info(msgId + ":消息发送成功");
mailSendLogService.updateMailSendLogStatus(msgId, 1);//修改数据库中的记录,消息投递成功
} else {
logger.info(msgId + ":消息发送失败");
}
});
rabbitTemplate.setReturnCallback((msg, repCode, repText, exchange, routingkey) -> {
logger.info("消息发送失败");
});
return rabbitTemplate;
}
//建立一个队列
@Bean
Queue mailQueue() {
return new Queue(MailConstants.MAIL_QUEUE_NAME, true);
}
//建立一个交换机
@Bean
DirectExchange mailExchange() {
return new DirectExchange(MailConstants.MAIL_EXCHANGE_NAME, true, false);
}
//把队列和交换机进行绑定,把交换器中的消息传输到绑定的队列实现消息消费
@Bean
Binding mailBinding() {
return BindingBuilder.bind(mailQueue()).to(mailExchange()).with(MailConstants.MAIL_ROUTING_KEY_NAME);
}
}
- 这个配置类被装配到其他和rabbit相关的代码块
- 通过传入cachingConnectionFactory,实例一个rabbit模板
- 判断消息是否传入成功
6.以上都包含于vhrserver这个包中,其中还有一个消费,即把传入的employee信息从交换器传入到Queue中实现消息的消费,在这个项目中通过发送邮件实现
编写了配置文件,配置qq邮箱的信息,配置消息中间件RabbitMq信息、还有redis等信息
server.port=8082
spring.mail.host=smtp.qq.com
spring.mail.protocol=smtp
spring.mail.default-encoding=UTF-8
spring.mail.password=**************
[email protected]
spring.mail.port=587
spring.mail.properties.mail.stmp.socketFactory.class=javax.net.ssl.SSLSocketFactory
spring.mail.properties.mail.debug=true
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
# spring.rabbitmq.host=192.168.91.128
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.prefetch=100
#spring.redis.host=192.168.91.128
spring.redis.host=localhost
spring.redis.port=6379
#spring.redis.password=123
spring.redis.database=0
利用thymeleaf编写了一个邮件模板
<!DOCTYPE html>
<html lang="en" xmlns:th="http://www.thymeleaf.org">
<head>
<meta charset="UTF-8">
<title>入职欢迎邮件</title>
</head>
<body>
欢迎 <span th:text="${name}"></span> 加入 Java达摩院 大家庭,您的入职信息如下:
<table border="1">
<tr>
<td>姓名</td>
<td th:text="${name}"></td>
</tr>
<tr>
<td>职位</td>
<td th:text="${posName}"></td>
</tr>
<tr>
<td>职称</td>
<td th:text="${joblevelName}"></td>
</tr>
<tr>
<td>部门</td>
<td th:text="${departmentName}"></td>
</tr>
</table>
<p>希望在未来的日子里,携手共进!</p>
</body>
</html>
实现邮件的发送
//对相应队列进行监听
@RabbitListener(queues = MailConstants.MAIL_QUEUE_NAME)
public void handler(Message message, Channel channel) throws IOException {
Employee employee = (Employee) message.getPayload();
MessageHeaders headers = message.getHeaders();
Long tag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
String msgId = (String) headers.get("spring_returned_message_correlation");
if (redisTemplate.opsForHash().entries("mail_log").containsKey(msgId)) {
//redis 中包含该 key,说明该消息已经被消费过
logger.info(msgId + ":消息已经被消费");
channel.basicAck(tag, false);//确认消息已消费
return;
}
//收到消息,发送邮件
MimeMessage msg = javaMailSender.createMimeMessage();
MimeMessageHelper helper = new MimeMessageHelper(msg);
try {
helper.setTo(employee.getEmail());
helper.setFrom(mailProperties.getUsername());
helper.setSubject("入职欢迎");
helper.setSentDate(new Date());
Context context = new Context();
context.setVariable("name", employee.getName());
context.setVariable("posName", employee.getPosition().getName());
context.setVariable("joblevelName", employee.getJobLevel().getName());
context.setVariable("departmentName", employee.getDepartment().getName());
String mail = templateEngine.process("mail", context);
helper.setText(mail, true);
javaMailSender.send(msg);
redisTemplate.opsForHash().put("mail_log", msgId, "javaboy");
channel.basicAck(tag, false);
logger.info(msgId + ":邮件发送成功");
} catch (MessagingException e) {
channel.basicNack(tag, false, true);
e.printStackTrace();
logger.error("邮件发送失败:" + e.getMessage());
}
}
参考