SpringBoot、RabbitMQ實作QQ郵件發送

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());
        }
    }

參考

SpringBoot、RabbitMQ實作QQ郵件發送

RabbitMQ郵件發送

RabbitMQ提供了4種不同的Exchange政策，分别是Direct，Fanout，Topic，Header

1.關于郵件的有兩個實體類

2.在業務層建立了MailSendLogService

3.對于MailSendLogService沒有對應的Controller層，是以，在Service層建立了一個MailSendTask，這個類用于處理沒發送成功

4.怎樣利用RabbitMQ作為消息中間件實作注冊消息的存儲和轉發

5.來看看RabbitConfig

6.以上都包含于vhrserver這個包中，其中還有一個消費，即把傳入的employee資訊從交換器傳入到Queue中實作消息的消費，在這個項目中通過發送郵件實作

編寫了配置檔案，配置qq郵箱的資訊，配置消息中間件RabbitMq資訊、還有redis等資訊

利用thymeleaf編寫了一個郵件模闆

實作郵件的發送

繼續閱讀

SSM整合02---Spring-SpringMVC層一、添加web架構支援二、web.xml三、spring-mvc.xml

SSM架構（二）------------表現層的SpringMVC

Intellij idea基于Maven 的SSM 架構整合

RabbitMQ指令行的基本操作

資料遷移方法資料遷移原則資料遷移之雙寫方案資料遷移之級聯同步方案

微服務-性能壓測\緩存redis和分布式鎖redisson和SpringCache

Nacos 2.0 更新前後性能對比壓測

Spring資料和Redis

redis叢集資料一緻性_RedisRaft為Redis叢集帶來強大的資料一緻性

SpringMVC的啟動過程源碼分析1 SpringMVC 配置XML的方式啟動2 SpringMVC純注解模式啟動

Java Spring 架構

java 8 Lamda 求 List 中 Bigdecimal類型的各種值求最大值求最小值求和求平均值

supervisor 管理redis 和httpd 環境centos7

Redis簡介一(單機版)發展曆程Redis

(SpringBoot)日志種類：log、monitor、access、out、gc、backup

SpringCloud微服務（四）之ribbon筆記ribbon是什麼？入門案例自定義均衡負載算法