把LangChain跑起來的三個方法
使用LangChain開發LLM應用時，需要機器進行GLM部署，好多同學第一步就被勸退了，那么如何繞過這個步驟先學習LLM模型的應用，對Langchain進行快速上手？本片講解3個把LangChain跑起來
本篇文章將介紹另一種高可用的服務架構，以便實現郵件 100% 被投遞成功。類似的短信推送等服務，實現邏輯也大體類似。
圖片
本文內容主要圍繞這個流程圖展開，利用 RabbitMQ 消息隊列來實現郵件 100% 被投遞，內容涵蓋了 RabbitMQ 很多知識點，如：
安裝 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=testspring.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=manualspring.rabbitmq.listener.simple.prefetch=100其中，spring.mail.username和spring.mail.password指的就是上文中創建的郵箱賬號和授權碼，將其配置進去即可。
編寫一個 RabbitMQ 配置類，用于監聽消息的發送情況，示例如下。
@Configurationpublic 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; }}在 Spring Boot 中，我們可以利用RabbitTemplate工具，將數據通過交換器發送到目標消息隊列，示例如下。
@Servicepublic 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; }}在 Spring Boot 中，我們可以利用@RabbitListener注解，監聽指定的消息隊列，如果隊列中有消息會第一時間收到回調，示例如下。
@Componentpublic 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); } }}正如之前的文章中所介紹的，在 Spring Boot 中，我們可以利用JavaMailSender工具來實現郵件的自動推送，示例如下。
@Servicepublic 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; } }}接著，編寫一個 controller 接口，將郵件發送服務暴露出去，示例如下：
@RestControllerpublic class MailController { @Autowired private ProduceService produceService; @PostMapping("send") public String sendMail(Mail mail) { boolean result = produceService.sendByAck(mail); return result ? "success": "fail"; }}最后，啟動 SpringBoot 服務，用 postman 來測試一下。
圖片
查看控制臺信息。
圖片
查詢接受者郵件信息。
圖片
可以清楚的看到，郵件發送成功！
當大批量的發送郵件，也不用擔心，因為整個郵件的發送都是異步的，不會阻塞主流程的運行。
雖然以上的方案非常可靠，可以保證發出的消息 100% 被消費，但是其實也有弊端。
試想一下，按照上面的處理邏輯，假設其中有一條消息，因為某種原因一直發送失敗，會出現什么樣的情況？
此時，這條消息會重新返回隊列，然后一直重試，會導致其它的消息可能會無法被消費。
針對這種情況，最簡單粗暴的辦法就是，當重試失敗之后將消息丟棄，不會阻礙其它的消息被正常處理，不過會丟失數據。
那么如何正確的處理消息消費失敗的問題呢？
可以借助數據庫來記錄消費失敗的數據，針對系統無法成功處理的消息，人工進行干預。
實踐過程如下！
首先，在數據庫中創建一張消息日志表，用于跟蹤消息數據的狀態，示例如下：
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消息日志';在生產者服務類中，先將消息數據寫入數據庫，再向 rabbitMQ 服務中發消息，示例如下：
@Servicepublic 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; }}在消費者服務類中，收到消息之后，不管處理成功還是失敗，都只會修改數據庫中的消息狀態，并且消息處理失敗時，不再重新返回隊列。
@Componentpublic 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當消息消費失敗時，會自動記錄到數據庫。
實際上，不可能每條數據都需要我們進行干預，有的可能重試一次就好了，因此可以編寫一個定時任務，將消費失敗的數據篩選出來，重新放入到消息隊列中，只有當消費次數達到設置的最大值，此時進入人工干預階段，可以節省不少的工作。
示例如下：
@Componentpublic 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@SpringBootApplicationpublic 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
本文鏈接：http://www.tebozhan.com/showinfo-26-112786-0.htmlSpringBoot整合RabbitMQ實現郵件異步發送
聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯系：2376512515@qq.com 津ICP備18001702號
津公網安備 12010102000574號