PHP基于Redis實作輕量級延遲佇列-有解無憂

延遲佇列，顧名思義它是一種帶有延遲功能的訊息佇列，那么，是在什么場景下我才需要這樣的佇列呢？

一、背景

先看看一下業務場景：

1.會員過期前3天發送召回通知
2.訂單支付成功后，5分鐘后檢測下游環節是否都正常，比如用戶購買會員后，各種會員狀態是否都設定成功
3.如何定期檢查處于退款狀態的訂單是否已經退款成功？
4.實作通知失敗，1，3，5，7分鐘重復通知，直到對方回復？

通常解決以上問題，最簡單直接的辦法就是定時去掃表，

掃表存在的問題是：

1.掃表與資料庫長時間連接，在數量量大的情況容易出現連接例外中斷，需要更多的例外處理，對程式健壯性要求高
2.在資料量大的情況下延時較高，規定內處理不完，影響業務，雖然可以啟動多個行程來處理，這樣會帶來額外的維護成本，不能從根本上解決，
3.每個業務都要維護一個自己的掃表邏輯，當業務越來越多時，發現掃表部分的邏輯會重復開發，但是非常類似

延時佇列能對于上述需求能很好的解決

二、調研

調研了市場上一些開源的方案，以下：

1.有贊科技：只有原理，沒有開源代碼
2.github個人的：https://github.com/ouqiang/delay-queue
1.基于redis實作，redis只能配置一個,如果redis掛了整個服務不可用，可用性差點
2.消費端實作的是拉模式，接入成本大，每個專案都得去實作一遍接入代碼
3.在star使用的人數不多，放在生產環境，存在風險，加之對go語言不了解，出了問題難以維護
3.SchedulerX-阿里開源的：功能很強大，但是運維復雜，依賴組件多，不夠輕量
4.RabbitMQ-延時任務: 本身沒有延時功能，需要借助一特性自己實作，而且公司沒有部署這個佇列，去單獨部署一個這個來做延時佇列成本有點高，而且還需要專門的運維來維護，目前團隊不支持

基本以上原因打算自己寫一個，平常使用php多，專案基本redis的zset結構作為存盤，用php語言實作，實作原理參考了有贊團隊：https://tech.youzan.com/queuing_delay/

整個延遲佇列主要由4個部分

JobPool用來存放所有Job的元資訊，
DelayBucket是一組以時間為維度的有序佇列，用來存放所有需要延遲的Job（這里只存放Job Id），
Timer負責實時掃描各個Bucket，并將delay時間大于等于當前時間的Job放入到對應的Ready Queue，
ReadyQueue存放處于Ready狀態的Job（這里只存放JobId），以供消費程式消費，
- - 　
    
    訊息結構
    每個Job必須包含一下幾個屬性：
    1. topic：Job型別，可以理解成具體的業務名稱，
    2. id：Job的唯一標識，用來檢索和洗掉指定的Job資訊，
    3. delayTime：jod延遲執行的時間，13位時間戳
    4. ttr（time-to-run)：Job執行超時時間，
    5. body：Job的內容，供消費者做具體的業務處理，以json格式存盤，
    對于同一類的topic delaytime,ttr一般是固定，job可以在精簡一下屬性
    
    1.topic：Job型別，可以理解成具體的業務名稱
    
    2.id：Job的唯一標識，用來檢索和洗掉指定的Job資訊，
    
    3.body：Job的內容，供消費者做具體的業務處理，以json格式存盤，
    
    delaytime,ttr在topicadmin后臺配置
    
    三、目標
    - 輕量級：有較少的php的拓展就能直接運行，不需要引入網路框架，比如swoole，workman之類的
    - 穩定性：采用master-work架構，master不做業務處理，只負責管理子行程，子行程例外退出時自動拉起
    - 可用性：
    - 拓展性: 當消費行程存在瓶頸時，可以配置加大消費行程數，當寫入存在瓶頸時，可增加實體數寫入性能可線性提高
    - 實時性：允許存在一定的時間誤差，
    - 支持訊息洗掉：業務使用方，可以隨時洗掉指定訊息，
    - 訊息傳輸可靠性：訊息進入到延遲佇列后，保證至少被消費一次，
    - 寫入性能：qps>1000+
    四、架構設計與說明
    
    總體架構
  - 采用master-work架構模式，主要包括6個模塊：
    - 1.dq-mster: 主行程，負責管理子行程的創建，銷毀，回收以及信號通知
    - 2.dq-server: 負責訊息寫入，讀取，洗掉功能以及維護redis連接池
    - 3.dq-timer-N: 負責從redis的zset結構中掃描到期的訊息，并負責寫入ready 佇列，個數可配置，一般2個就行了，因為訊息在zset結構是按時間有序的
    - 4.dq-consume-N: 負責從ready佇列中讀取訊息并通知給對應回呼介面，個數可配置
    - 5.dq-redis-checker: 負責檢查redis的服務狀態，如果redis宕機，發送告警郵件
    - 6.dq-http-server: 提供web后臺界面，用于注冊topic
    五、模塊流程圖
    
    訊息寫入:
  - timer查找到期訊息:
  - timer查找到期訊息:
    
    consumer消費流程:
    
    六、部署
    
    環境依賴：PHP 5.4+ 安裝sockets，redis，pcntl,pdo_mysql 拓展
    
    ps: 熟悉docker的同學可以直接用鏡像: shareclz/php7.2.14 里面包含了所需拓展
    
    step1:安裝資料庫用于存盤一些topic以及告警資訊
    
    執行:
    
    mysql> source dq.sql
    
    step2:在DqConfg.檔案中配置資料庫資訊： DqConf::$db step3: 啟動http服務
    在DqConf.php檔案中修改php了路徑
    
    命令：
    
    php DqHttpServer.php --port 8088
    
    訪問:http://127.0.0.1:8088,出現配置界面
    
    redis資訊格式：host:port:auth 比如 127.0.0.1:6379:12345
    
    stop4:配置告資訊(比如redis宕機)
    
    stop5:注冊topic
    
    重試標記說明:
    
    1.介面回傳為空默認重試
    
    2.滿足指定回傳表達會重試，res表示回傳的json陣列，比如:
    
    回呼介面回傳json串：{ "code": 200, "data":{ "status": 2, "msg": "回傳失敗"}}，重試條件可以這樣寫
    
    {res.code}!= 200
    
    {res.code}!= 200&& {res. data.status}!= 2
    
    {res.code}== 200&& {res. data.status}== 2|| {res. data.msg}== '回傳失敗'
    
    step6:啟動服務行程:
    
    php DqInit.php --port 6789 &
    
    執行 ps -ef | grep dq 看到如下資訊說明啟動成功
    
    step7: 寫入資料，參考demo.phpstep8:查看日志
    
    默認日志目錄在專案目錄的logs目錄下，在DqConf.php修改$logPath
    - 1.請求日志：request_ymd.txt
    - 2.通知日志：notify_ymd.txt
    - 3.錯誤日志：err_ymd.txt
    step9:如果組態檔有改動
    - 1.系統會自動檢測組態檔新，如果有改動，會自動退出(沒有找到較好的熱更新的方案)，需要重啟，可以在crontab里面建個任務,1分鐘執行一次，程式有check_self的判斷
    - 2.優雅退出命令: master檢測偵聽了USR2信號，收到信號后會通知所有子行程，子行程完成當前任務后會自動退出
    ps -ef | grep dq-master| grep -v grep | head -n 1 | awk '{print $2}' | xargs kill -USR2
    
    七、性能測驗
    需要安裝pthreads拓展：
    
    測驗原理：使用多執行緒模擬并發，在1s內能成功回傳請求成功的個數
    
    php DqBench concurrency requests
    
    concurrency:并發數
    
    requests：每個并發產生的請求數
    
    測驗環境：記憶體 8G ，8核cpu，2個redis和1個dq-server 部署在一個機器上，資料包64位元組
    
    qps：2400
    
    八、值得一提的性能優化點：
    - 1.redis multi命令：將多個對redis的操作打包成一個減少網路開銷
    - 2.計數的操作異步處理，在異步邏輯里面用函式的static變數來保存，當寫入redis成功后釋放static變數，可以在redis出現例外時計數仍能保持一致，除非行程退出
    - 3.記憶體泄露檢測有必要: 所有的記憶體分配在底層都是呼叫了brk或者mmap，只要程式只有大量brk或者mmap的系統呼叫，記憶體泄露可能性非常高 ,檢測命令: strace -c -p pid | grep -P 'mmap| brk'
    - 4.檢測程式的系統呼叫情況：strace -c -p pid ，發現某個系統函式呼叫是其他的數倍，可能大概率程式存在問題
    九、例外處理
    1.如果呼叫通知介面在超時時間內，沒有收到回復認為通知失敗，系統會重新把資料放入佇列，重新通知，系統默認最大通知10次(可以在Dqconf.php檔案中修改$notify_exp_nums)通知間隔為2n+1，比如第一次1分鐘，通知失敗，第二次3分鐘后，直到收到回復，超出最大通知次數后系統自動丟棄，同時發郵件通知
    
    2.線上redis每隔1s持久化一次，存在丟失1s資料的情況，出現這種情況可以對比request_ymd.txt和notify_ymd.txt日志手動恢復過來
    
    3.redis宕機通知:
    
    ps:網路抖動在所難免，通知介面如果涉及到核心的服務,一定要保證冪等！！
    
    十、線上情況
    
    線上部署了兩個實體每個機房部一個，4個redis共16G記憶體作存盤，服務穩定運行數月，各項指標均符合預期
    
    主要接入業務:
    - 訂單10分鐘召回通知
    - 呼叫介面超時或者失敗時做補償
    - 會員過期前3天召回通知
    十一、不足與展望
    1.由于團隊使用的鏡像缺少libevent拓展，所以dq-server基于select模型，并發高的場景下性能存在瓶頸，后續可以改為基于libevent事件模型，提升并發性能
    
    2.timer和consumer目前是采用多行程來做的，這個粒度感覺有點粗，可以考慮使用多執行緒模式，并且支持動態創建執行緒數來提高consumer的性能，最大程度保證消費及時
    
    3.dq-server與redis是同步呼叫，這也是性能的瓶頸點，計劃基于swoole_redis來異步處理
  - 推薦閱讀：
  - Swoole和Redis實作的并發佇列處理系統