發(fā)送千萬級(jí)消息，我是這樣設(shè)計(jì)的

我是3y，一年CRUD經(jīng)驗(yàn)用十年的markdown程序員???????常年被譽(yù)為優(yōu)質(zhì)八股文選手

前幾天，我講了什么是定時(shí)任務(wù)、為什么要用定時(shí)任務(wù)、什么是分布式定時(shí)任務(wù)、分布式定時(shí)任務(wù)的基礎(chǔ)以及austin項(xiàng)目接入分布式定時(shí)任務(wù)的背景。

有幾個(gè)同學(xué)表示之前沒接觸過分布式定時(shí)任務(wù)，想讓我來austin的相關(guān)邏輯，于是就有了這篇文章。

01、使用定時(shí)任務(wù)推送消息

對(duì)于austin消息推送平臺(tái)而言，發(fā)送消息不單單是「技術(shù)側(cè)」調(diào)用接口進(jìn)行發(fā)送的，還有很多場(chǎng)景是「運(yùn)營(yíng)側(cè)」通過設(shè)置定時(shí)進(jìn)而推送。

所以，作為使用消息推送平臺(tái)的角色可以簡(jiǎn)單分為：「技術(shù)同學(xué)」、「運(yùn)營(yíng)同學(xué)」

當(dāng)運(yùn)營(yíng)或客服她們想發(fā)送消息給用戶，她們使用消息推送的后臺(tái)，步驟可簡(jiǎn)單分為：

1、確定發(fā)送時(shí)間

2、確定發(fā)送的人群（人群內(nèi)可以是1~N的用戶）

3、確定發(fā)送文案

對(duì)于發(fā)送時(shí)間則使用cron表達(dá)式，發(fā)送人群我們是讓運(yùn)營(yíng)上傳.csv文件，發(fā)送文案則配置在模板上（同樣也可以使用占位符）

這里值得講述的是，為什么是上傳.csv文件而不是excel，其最根本的原因：excel的行數(shù)是有大小限制的，而.csv的行數(shù)是沒有大小限制的。

我們限定.csv文件的格式為如下：第一列填寫接收者Id、剩余的列如果使用了占位符，則需要填寫占位符變量名

保存了消息模板之后，等我們點(diǎn)擊「啟動(dòng)」按鈕，就根據(jù)模板的信息進(jìn)行消息推送。

（在線上環(huán)境上，在啟動(dòng)之前肯定會(huì)有審核的環(huán)節(jié)，并且一般會(huì)先點(diǎn)擊「測(cè)試」按鈕看文案是否正常才進(jìn)行推送）

點(diǎn)擊啟動(dòng)按鈕了之后，會(huì)發(fā)生什么？我們看日志就懂了（我在執(zhí)行的過程中把關(guān)鍵的日志信息都打印出來了）

//1.?消息模板ID的消息?定時(shí)任務(wù)被觸發(fā)（入口）
2022-02-17?21:14:55.016?[Thread-61]?INFO??com.java3y.austin.cron.handler.CronTaskHandler?-?CronTaskHandler#execute?messageTemplateId:7?cron?exec!

//2.?api-service接口接收到的參數(shù)信息以及接口返回值???
2022-02-17?21:14:56.095?[pool-27-thread-1]?INFO??com.java3y.austin.support.utils.LogUtils?-?{"bizId":"7","bizType":"SendService#batchSend","executionTime":72,"logId":"9fd2cdd1-79fa-4c5c-932b-da08aa3b247d","msg":"{\"code\":\"send\",\"messageParamList\":[{\"extra\":null,\"receiver\":\"13719383334,13719383336,13719383338,13719383340,13719383342,13719383344\",\"variables\":{\"content\":\"xixi\",\"url\":\"hehe.com\"}},{\"extra\":null,\"receiver\":\"13719383333,13719383335,13719383337,13719383339,13719383341,13719383343,13719383345\",\"variables\":{\"content\":\"hhaha\",\"url\":\"baidu.com\"}}],\"messageTemplateId\":7}","operateDate":1645103696095,"returnStr":"{\"code\":\"0\",\"msg\":\"操作成功\"}","success":true,"tag":"operation"}

//3.?receiver消息隊(duì)列接收到的原始值
2022-02-17?21:14:56.252?[org.springframework.kafka.KafkaListenerEndpointContainer#6-0-C-1]?INFO??com.java3y.austin.support.utils.LogUtils?-?{"bizType":"Receiver#consumer","object":{"businessId":1000000720220217,"contentModel":{"content":"xixi","url":"hehe.com?track_code_bid=1000000720220217"},"idType":30,"messageTemplateId":7,"msgType":10,"receiver":["13719383340","13719383336","13719383338","13719383342","13719383344","13719383334"],"sendAccount":10,"sendChannel":30,"templateType":10},"timestamp":1645103696252}
2022-02-17?21:14:56.254?[org.springframework.kafka.KafkaListenerEndpointContainer#6-0-C-1]?INFO??com.java3y.austin.support.utils.LogUtils?-?{"bizType":"Receiver#consumer","object":{"businessId":1000000720220217,"contentModel":{"content":"hhaha","url":"baidu.com?track_code_bid=1000000720220217"},"idType":30,"messageTemplateId":7,"msgType":10,"receiver":["13719383341","13719383339","13719383335","13719383337","13719383343","13719383333","13719383345"],"sendAccount":10,"sendChannel":30,"templateType":10},"timestamp":1645103696253}

//4.1?關(guān)鍵位置打印的日志(state=10)代表消息隊(duì)列接收成功
2022-02-17?21:14:56.172?[org.springframework.kafka.KafkaListenerEndpointContainer#6-0-C-1]?INFO??com.java3y.austin.support.utils.LogUtils?-?{"businessId":1000000720220217,"ids":["13719383340","13719383336","13719383338","13719383342","13719383344","13719383334"],"state":10,"timestamp":1645103696172}
2022-02-17?21:14:56.253?[org.springframework.kafka.KafkaListenerEndpointContainer#6-0-C-1]?INFO??com.java3y.austin.support.utils.LogUtils?-?{"businessId":1000000720220217,"ids":["13719383341","13719383339","13719383335","13719383337","13719383343","13719383333","13719383345"],"state":10,"timestamp":1645103696253}

//4.2?關(guān)鍵位置打印的日志(state=60)代表消息調(diào)用接口發(fā)送失敗
2022-02-17?21:14:56.799?[pool-8-thread-3]?INFO??com.java3y.austin.support.utils.LogUtils?-?{"businessId":1000000720220217,"ids":["13719383340","13719383336","13719383338","13719383342","13719383344","13719383334"],"state":60,"timestamp":1645103696799}
??

因?yàn)檫@次發(fā)送的渠道是短信，我們是有將短信的發(fā)送記錄入庫(kù)的，所以可以看看數(shù)據(jù)庫(kù)的記錄（剛好是13條，數(shù)據(jù)是沒問題的。至于發(fā)失敗，主要是該短信模板的參數(shù)不符合）

我們整體的流程是沒有問題的

02、代碼結(jié)構(gòu)設(shè)計(jì)

從消息推送后臺(tái)層面上，當(dāng)點(diǎn)擊了「啟動(dòng)」按鈕時(shí)，其實(shí)是「創(chuàng)建&&啟動(dòng)」或者「啟動(dòng)」了個(gè)定時(shí)任務(wù)（調(diào)用xxl-job的api，將定時(shí)任務(wù)存入到xxl-job的數(shù)據(jù)庫(kù)表中）

等我們定時(shí)任務(wù)到時(shí)間點(diǎn)了，xxl-job的調(diào)度中心就找到我們的執(zhí)行器，進(jìn)行調(diào)用。

在創(chuàng)建定時(shí)任務(wù)的時(shí)候，我把消息模板ID寫入到了xxl-job任務(wù)信息里，所以當(dāng)任務(wù)被調(diào)度的時(shí)候，我又從xxl-job把參數(shù)信息取出來。在這，打印出了一條日志，表示當(dāng)前模板ID被調(diào)度執(zhí)行了。

隨后，我使用「線程池」對(duì)定時(shí)任務(wù)做處理。

因?yàn)槲疫@里認(rèn)為「讀取文件以及遠(yuǎn)程調(diào)用發(fā)送接口」是一件比較耗時(shí)的工作，所以我這里直接就用線程池做了層異步，及時(shí)返回xxl-job，避免定時(shí)任務(wù)超時(shí)。

解釋完為什么用了線程池以后，接著來看看讀取.csv文件這塊。在最最最開始的時(shí)候，我是直接一次性讀取，然后得到List列表的。顯然，這是不合理的。假設(shè)運(yùn)營(yíng)圈選的人群可能達(dá)到2000W人，那我直接將2000W條記錄直接load到內(nèi)存，那是不對(duì)的。

所以，每當(dāng)從文件讀取一行，我就處理一行

我在拿到每一行數(shù)據(jù)的時(shí)候，封裝了一個(gè)VO，又扔給了內(nèi)存隊(duì)列LazyPending

內(nèi)存隊(duì)列里會(huì)起一個(gè)線程消費(fèi)隊(duì)列里的數(shù)據(jù)，等到積壓到給定的size或者timeout就會(huì)給到實(shí)際消費(fèi)者進(jìn)行處理

我這樣設(shè)計(jì)的目的在于：我要調(diào)用批量發(fā)送接口，使用內(nèi)存隊(duì)列作為介質(zhì)實(shí)現(xiàn)生產(chǎn)者和消費(fèi)者模式為了做batch處理。具體來說：如果我每讀取一行就調(diào)用一次發(fā)送接口，假設(shè)人群有2000W，我就需要調(diào)用2000W次。

在實(shí)際生產(chǎn)環(huán)境中，austin-cron和austin-api在很大概率上是分開部署的，所以每一次調(diào)用接口都是遠(yuǎn)程調(diào)用。為了減少這個(gè)消耗，所以我這樣干了。

另外，在具體執(zhí)行消費(fèi)的時(shí)候，我是設(shè)計(jì)了「線程池」進(jìn)行接口調(diào)用的，更能充分利用系統(tǒng)資源（畢竟這次接口調(diào)用更多的是損耗網(wǎng)絡(luò)的開銷）

看到了這，再回到我們的接口打印信息：

2022-02-17?21:14:56.095?[pool-27-thread-1]?INFO??com.java3y.austin.support.utils.LogUtils?-?{"bizId":"7","bizType":"SendService#batchSend","executionTime":72,"logId":"9fd2cdd1-79fa-4c5c-932b-da08aa3b247d","msg":"{\"code\":\"send\",\"messageParamList\":[{\"extra\":null,\"receiver\":\"13719383334,13719383336,13719383338,13719383340,13719383342,13719383344\",\"variables\":{\"content\":\"xixi\",\"url\":\"hehe.com\"}},{\"extra\":null,\"receiver\":\"13719383333,13719383335,13719383337,13719383339,13719383341,13719383343,13719383345\",\"variables\":{\"content\":\"hhaha\",\"url\":\"baidu.com\"}}],\"messageTemplateId\":7}","operateDate":1645103696095,"returnStr":"{\"code\":\"0\",\"msg\":\"操作成功\"}","success":true,"tag":"operation"}

人群數(shù)量一共是13，但我們僅用了一次接口調(diào)用。

03、總結(jié)

到這里，我已經(jīng)把定時(shí)任務(wù)的實(shí)現(xiàn)核心邏輯應(yīng)該就講完了，大家看代碼的時(shí)候應(yīng)該就有個(gè)譜了，至少應(yīng)該不會(huì)跑到群里說看不懂了。

至于實(shí)現(xiàn)的細(xì)節(jié)上，如果有更好的辦法或者思路可以在評(píng)論區(qū)一起討論，又或是直接提個(gè)PR。一般我認(rèn)為是合理的，我都會(huì)審核通過的喲！

看完整篇文章，很有可能就會(huì)有同學(xué)有疑惑：你把數(shù)據(jù)放在內(nèi)存隊(duì)列里，這如果重啟或系統(tǒng)掛了怎么辦啊，數(shù)據(jù)不就丟了嗎。其實(shí)我認(rèn)為這是一種權(quán)衡traff-off。

我們?cè)谙到y(tǒng)里要保證數(shù)據(jù)不丟失不重復(fù)需要做大量的工作，很有可能會(huì)影響到系統(tǒng)的性能或者支持并發(fā)的大小。如果是處理訂單類的系統(tǒng)，那是必須的。但如果是發(fā)消息的場(chǎng)景，或者并沒有想象中那么重要（當(dāng)然了，我們也可以實(shí)現(xiàn)就是啦）。

但更多的是，我們可以額外通過一些手段來判斷消息是否下發(fā)成功了：大概就是統(tǒng)計(jì)當(dāng)前消息模板的下發(fā)人數(shù)、系統(tǒng)處理過程中的人數(shù)以及消息到達(dá)、點(diǎn)擊的人數(shù)。這是系統(tǒng)核心功能以外的，但又很重要的功能，這幾天已經(jīng)在實(shí)現(xiàn)了，這周有望寫完。

《對(duì)線面試官》公眾號(hào)還在持續(xù)分享面試題，沒關(guān)注的同學(xué)可以關(guān)注一波！這是austin項(xiàng)目的上一個(gè)系列，質(zhì)量桿桿的

austin項(xiàng)目Gitee鏈接：https://gitee.com/zhongfucheng/austin

austin項(xiàng)目GitHub鏈接：https://github.com/ZhongFuCheng3y/austin

閱讀原文可跳轉(zhuǎn)austin倉(cāng)庫(kù)