SpringBoot整合RabbitMQ實現郵件異步發送

作者：志哥聊技術 2024-09-05 08:58:37

開發架構

本文主要以實現郵件自動推送這個業務場景為例，通過 Springboot 整合 rabbitMQ 技術來實現高可用的效果。

本篇文章將介紹另一種高可用的服務架構，以便實現郵件 100% 被投遞成功。類似的短信推送等服務，實現邏輯也大體類似。

01、先來一張流程圖

圖片

本文內容主要圍繞這個流程圖展開，利用 RabbitMQ 消息隊列來實現郵件 100% 被投遞，內容涵蓋了 RabbitMQ 很多知識點，如：

生產者和消費者模型
消息發送機制
消費確認機制
消息的重新投遞
消息消費失敗的處理方案

02、實現思路

1.準備一臺電腦，并安裝 RabbitMQ 服務
2.開放 QQ 郵箱或者其它郵箱授權碼，用于發送郵件
3.創建郵件發送項目并編寫代碼
4.發送郵件測試
5.消息消費失敗的處理介紹

03、環境準備

3.1、安裝 RabbitMQ 服務

安裝 RabbitMQ 服務，這一步比較簡單，可以訪問下面的官方地址，下載軟件包并依次按照步驟進行安裝即可。

https://rabbitmq.org.cn/docs/download

安裝成功之后，登陸 RabbitMQ 控制臺，可以看到類似于如下界面。

圖片

3.1.1、創建交換器

點擊“Exchanges”菜單，進入“交換器”管理界面。

圖片

進入之后，點擊最下方“Add a new exchange”按鈕，創建一個類型為topic，名稱叫mail.exchange的交換器，并提交。

圖片

3.1.2、創建消息隊列

接著，點擊“Queues”菜單，進入消息隊列管理界面。

圖片

同樣的，點擊最下方“Add a new queue”按鈕，創建一個名稱叫mq.mail.ack的消息隊列，并提交。

圖片

保存之后，在列表中可以看到剛剛創建的消息隊列，然后點擊進入詳情。

圖片

在詳情中，將當前消息隊列與上文創建的交換器進行綁定，便于后續通過交換器來發送消息到隊列，操作如下。

圖片

對于topic類型的交換器，通常不直接與消息隊列進行交互，而是通過一個路由鍵，將消息路由到目標消息隊列，這樣設計的目的是讓消息投遞更加靈活。路由鍵，可以簡單理解為類似于路由器，對數據進行路由分發處理。

3.2、配置郵箱發送服務器

為了實現郵件自動發送功能，我們還需要準備一個郵箱發送服務器，這一步在之前的文章中已經詳細的介紹過，在此，我們簡單的再介紹一下。

以 QQ 郵箱為例，登陸進去之后，在設置里面開啟 POP3/SMTP 服務，并獲取授權碼記錄下來。

圖片

該授權碼，就是下文配置文件中spring.mail.password需要的密碼！

04、方案實踐

4.1、構建項目

在 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>

4.2、添加相關配置

在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=true
spring.mail.properties.mail.smtp.ssl.enable=true
spring.mail.properties.mail.smtp.ssl.required=true

#rabbitmq配置
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=test
spring.rabbitmq.password=test
# 開啟confirms回調 P -> Exchange
spring.rabbitmq.publisher-cnotallow=true
# 開啟returnedMessage回調 Exchange -> Queue
spring.rabbitmq.publisher-returns=true
# 設置手動確認(ack) Queue -> C
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.prefetch=100

其中，spring.mail.username和spring.mail.password指的就是上文中創建的郵箱賬號和授權碼，將其配置進去即可。

4.3、編寫 RabbitMQ 配置類

編寫一個 RabbitMQ 配置類，用于監聽消息的發送情況，示例如下。

@Configuration
public class RabbitConfig {

    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitConfig.class);


    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 設置消息轉換器為json格式
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());

        // 消息是否成功發送到Exchange
        rabbitTemplate.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);
        });

        return rabbitTemplate;
    }
}

4.4、編寫生產者服務

在 Spring Boot 中，我們可以利用RabbitTemplate工具，將數據通過交換器發送到目標消息隊列，示例如下。

@Service
public class ProduceService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 發送消息
     * @param mail
     * @return
     */
    public boolean sendByAck(Mail mail) {
        // 創建uuid
        String 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);
        return true;
    }
}

4.5、編寫消費者服務

在 Spring Boot 中，我們可以利用@RabbitListener注解，監聽指定的消息隊列，如果隊列中有消息會第一時間收到回調，示例如下。

@Component
public class ConsumerService {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerService.class);

    @Autowired
    private SendMailService sendMailService;

    /**
     * 監聽消息隊列，手動確認模式，必須手動調用ack或者nack方法
     * 配置參數：spring.rabbitmq.listener.simple.acknowledge-mode=manual
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(queues = {"mq.mail.ack"})
    public void consumeFromAck(Message message, Channel channel) throws IOException {
        LOGGER.info("收到消息：{}", message.toString());
        //將消息轉化為對象
        Mail mail = MessageHelper.msgToObj(message, Mail.class);

        // 手動確認模式
        long tag = message.getMessageProperties().getDeliveryTag();
        boolean success = sendMailService.send(mail);
        if (success) {
            // 消費成功，消息會被刪除
            channel.basicAck(tag, false);
        } else {
            // 消費失敗，重新返回隊列
            channel.basicNack(tag, false, true);
        }
    }
}

4.6、編寫郵件發送服務

正如之前的文章中所介紹的，在 Spring Boot 中，我們可以利用JavaMailSender工具來實現郵件的自動推送，示例如下。

@Service
public class SendMailService {

    private static final Logger LOGGER = LoggerFactory.getLogger(SendMailService.class);


    @Value("${spring.mail.username}")
    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);
            LOGGER.info("郵件發送成功");
            return true;
        } catch (MailException e) {
            LOGGER.error("郵件發送失敗, to: {}, title: {}", to, title, e);
            return false;
        }
    }
}

4.7、編寫 controller 接口

接著，編寫一個 controller 接口，將郵件發送服務暴露出去，示例如下：

@RestController
public class MailController {

    @Autowired
    private ProduceService produceService;

    @PostMapping("send")
    public String sendMail(Mail mail) {
        boolean result = produceService.sendByAck(mail);
        return result ? "success": "fail";
    }
}

4.8、服務測試

最后，啟動 SpringBoot 服務，用 postman 來測試一下。

圖片

查看控制臺信息。

圖片

查詢接受者郵件信息。

圖片

可以清楚的看到，郵件發送成功！

當大批量的發送郵件，也不用擔心，因為整個郵件的發送都是異步的，不會阻塞主流程的運行。

05、消費失敗的處理方案

雖然以上的方案非常可靠，可以保證發出的消息 100% 被消費，但是其實也有弊端。

試想一下，按照上面的處理邏輯，假設其中有一條消息，因為某種原因一直發送失敗，會出現什么樣的情況？

此時，這條消息會重新返回隊列，然后一直重試，會導致其它的消息可能會無法被消費。

針對這種情況，最簡單粗暴的辦法就是，當重試失敗之后將消息丟棄，不會阻礙其它的消息被正常處理，不過會丟失數據。

那么如何正確的處理消息消費失敗的問題呢？

可以借助數據庫來記錄消費失敗的數據，針對系統無法成功處理的消息，人工進行干預。

實踐過程如下！

5.1、創建一張消息日志表

首先，在數據庫中創建一張消息日志表，用于跟蹤消息數據的狀態，示例如下：

CREATE TABLE `msg_log` (
  `msg_id` varchar(255) NOT NULL DEFAULT '' COMMENT '消息唯一標識',
  `exchange` varchar(100) NOT NULL DEFAULT '' COMMENT '交換機',
  `route_key` varchar(100) NOT NULL DEFAULT '' COMMENT '路由鍵',
  `queue_name` varchar(100) NOT NULL DEFAULT '' COMMENT '隊列名稱',
  `msg` text COMMENT '消息體, json格式化',
  `result` varchar(255) DEFAULT NULL COMMENT '處理結果',
  `status` int(11) NOT NULL DEFAULT '0' COMMENT '狀態，0：等待消費，1：消費成功，2：消費失敗，9：重試失敗',
  `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`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='mq消息日志';

5.2、改寫生產者邏輯

在生產者服務類中，先將消息數據寫入數據庫，再向 rabbitMQ 服務中發消息，示例如下：

@Service
public class ProduceService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MsgLogService msgLogService;

    /**
     * 發送消息
     * @param mail
     * @return
     */
    public boolean sendByAuto(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);
        return true;
    }
}

5.3、改寫消費者邏輯

在消費者服務類中，收到消息之后，不管處理成功還是失敗，都只會修改數據庫中的消息狀態，并且消息處理失敗時，不再重新返回隊列。

@Component
public class ConsumerService {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerService.class);

    @Autowired
    private SendMailService sendMailService;

    @Autowired
    private MsgLogService msgLogService;

    /**
     * 監聽消息隊列，自動確認模式，無需調用ack或者nack方法，當程序執行時才刪除消息
     * 配置參數：spring.rabbitmq.listener.simple.acknowledge-mode=auto
     * @param message
     */
    @RabbitListener(queues = {"mq.mail.auto"})
    public void consumeFromAuto(Message message) {
        LOGGER.info("收到消息：{}", message.toString());
        // 獲取消息ID
        Mail mail = MessageHelper.msgToObj(message, Mail.class);

        // 消息冪等性處理，如果已經處理成功，無需重復消費
        MsgLog queryObj = msgLogService.selectByMsgId(mail.getMsgId());
        if(Objects.nonNull(queryObj) && Constant.SUCCESS.equals(queryObj.getStatus())){
            return;
        }

        // 發送郵件
        boolean success = 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

5.4、編寫定時任務對失敗消息進行補償投遞

當消息消費失敗時，會自動記錄到數據庫。

實際上，不可能每條數據都需要我們進行干預，有的可能重試一次就好了，因此可以編寫一個定時任務，將消費失敗的數據篩選出來，重新放入到消息隊列中，只有當消費次數達到設置的最大值，此時進入人工干預階段，可以節省不少的工作。

示例如下：

@Component
public class ScheduledTask {

    private static final Logger LOGGER = LoggerFactory.getLogger(ScheduledTask.class);

    /**
     * 最大投遞次數
     */
    private static final int MAX_TRY_COUNT = 3;

    @Autowired
    private MsgLogService msgLogService;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 每30s拉取消費失敗的消息, 重新投遞
     */
    @Scheduled(cron = "0/30 * * * * ?")
    public void 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% 消費成功！

06、小結