php實現(xiàn)延遲隊列

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

一、背景

先看看一下業(yè)務(wù)場景：

• 1.會員過期前3天發(fā)送召回通知
• 2.訂單支付成功后，5分鐘后檢測下游環(huán)節(jié)是否都正常，比如用戶購買會員后，各種會員狀態(tài)是否都設(shè)置成功
• 3.如何定期檢查處于退款狀態(tài)的訂單是否已經(jīng)退款成功？
• 4.實現(xiàn)通知失敗，1，3，5，7分鐘重復(fù)通知，直到對方回復(fù)？

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

掃表存在的問題是：

• 1.掃表與數(shù)據(jù)庫長時間連接，在數(shù)量量大的情況容易出現(xiàn)連接異常中斷，需要更多的異常處理，對程序健壯性要求高
• 2.在數(shù)據(jù)量大的情況下延時較高，規(guī)定內(nèi)處理不完，影響業(yè)務(wù)，雖然可以啟動多個進(jìn)程來處理，這樣會帶來額外的維護(hù)成本，不能從根本上解決。
• 3.每個業(yè)務(wù)都要維護(hù)一個自己的掃表邏輯。 當(dāng)業(yè)務(wù)越來越多時，發(fā)現(xiàn)掃表部分的邏輯會重復(fù)開發(fā)，但是非常類似

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

二、調(diào)研

調(diào)研了市場上一些開源的方案，以下：

• 1.有贊科技：只有原理，沒有開源代碼

• 2.github個人的：https://github.com/ouqiang/delay-queue

  1.基于redis實現(xiàn)，redis只能配置一個,如果redis掛了整個服務(wù)不可用，可用性差點
  2.消費(fèi)端實現(xiàn)的是拉模式，接入成本大，每個項目都得去實現(xiàn)一遍接入代碼
  3.在star使用的人數(shù)不多，放在生產(chǎn)環(huán)境，存在風(fēng)險，加之對go語言不了解，出了問題難以維護(hù)

• 3.SchedulerX-阿里開源的： 功能很強(qiáng)大，但是運(yùn)維復(fù)雜，依賴組件多，不夠輕量
• 4.RabbitMQ-延時任務(wù): 本身沒有延時功能，需要借助一特性自己實現(xiàn)，而且公司沒有部署這個隊列，去單獨部署一個這個來做延時隊列成本有點高，而且還需要專門的運(yùn)維來維護(hù)，目前團(tuán)隊不支持

基本以上原因打算自己寫一個，平常使用php多，項目基本redis的zset結(jié)構(gòu)作為存儲，用php語言實現(xiàn) ，實現(xiàn)原理參考了有贊團(tuán)隊：https://tech.youzan.com/queuing_delay/

三、目標(biāo)

• 輕量級：有較少的php的拓展就能直接運(yùn)行，不需要引入網(wǎng)絡(luò)框架，比如swoole，workman之類的
• 穩(wěn)定性：采用master-work架構(gòu)，master不做業(yè)務(wù)處理，只負(fù)責(zé)管理子進(jìn)程，子進(jìn)程異常退出時自動拉起
• 可用性：
  • 1.支持多實例部署，每個實例無狀態(tài)，一個實例掛掉不影響服務(wù)
  • 2.支持配置多個redis，一個redis掛了只影響部分消息
  • 3.業(yè)務(wù)方接入方便，在后臺只需填寫相關(guān)消息類型和回掉接口
• 拓展性: 當(dāng)消費(fèi)進(jìn)程存在瓶頸時，可以配置加大消費(fèi)進(jìn)程數(shù)，當(dāng)寫入存在瓶頸時，可增加實例數(shù)寫入性能可線性提高
• 實時性：允許存在一定的時間誤差。
• 支持消息刪除：業(yè)務(wù)使用方，可以隨時刪除指定消息。
• 消息傳輸可靠性：消息進(jìn)入到延遲隊列后，保證至少被消費(fèi)一次。
• 寫入性能：qps>1000+

四、架構(gòu)設(shè)計與說明

總體架構(gòu)

采用master-work架構(gòu)模式，主要包括6個模塊：

• 1.dq-mster: 主進(jìn)程，負(fù)責(zé)管理子進(jìn)程的創(chuàng)建，銷毀，回收以及信號通知
• 2.dq-server: 負(fù)責(zé)消息寫入，讀取，刪除功能以及維護(hù)redis連接池
• 3.dq-timer-N: 負(fù)責(zé)從redis的zset結(jié)構(gòu)中掃描到期的消息，并負(fù)責(zé)寫入ready 隊列，個數(shù)可配置，一般2個就行了，因為消息在zset結(jié)構(gòu)是按時間有序的
• 4.dq-consume-N: 負(fù)責(zé)從ready隊列中讀取消息并通知給對應(yīng)回掉接口，個數(shù)可配置
• 5.dq-redis-checker: 負(fù)責(zé)檢查redis的服務(wù)狀態(tài)，如果redis宕機(jī)，發(fā)送告警郵件
• 6.dq-http-server: 提供web后臺界面，用于注冊topic

五、部署

環(huán)境依賴：PHP 5.4+ 安裝sockets，redis，pcntl,pdo_mysql 拓展

create database dq;
#存放告警信息
CREATE TABLE `dq_alert` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `host` varchar(255) NOT NULL DEFAULT '',
  `port` int(11) NOT NULL DEFAULT '0',
  `user` varchar(255) NOT NULL DEFAULT '',
  `pwd` varchar(255) NOT NULL DEFAULT '',
  `ext` varchar(2048) NOT NULL DEFAULT '',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;
#存放redis信息
CREATE TABLE `dq_redis` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `t_name` varchar(200) NOT NULL DEFAULT '',
  `t_content` varchar(2048) NOT NULL DEFAULT '',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=14 DEFAULT CHARSET=utf8;
#存儲注冊信息
CREATE TABLE `dq_topic` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `t_name` varchar(1024) NOT NULL DEFAULT '',
  `delay` int(11) NOT NULL DEFAULT '0',
  `callback` varchar(1024) NOT NULL DEFAULT '',
  `timeout` int(11) NOT NULL DEFAULT '3000',
  `email` varchar(1024) NOT NULL DEFAULT '',
  `topic` varchar(255) NOT NULL DEFAULT '',
  `createor` varchar(1024) NOT NULL DEFAULT '',
  `status` tinyint(4) NOT NULL DEFAULT '1',
  `method` varchar(32) NOT NULL DEFAULT 'GET',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8;

在DqConf.php文件中修改php了路徑 $logPath

命令：

php DqHttpServer.php --port 8088

訪問:http://127.0.0.1:8088,出現(xiàn)配置界面

redis信息格式：host:post:auth 比如 127.0.0.1:6379:12345

php DqInit.php --port 6789
看到如下信息說明啟動成功

<?php
include_once 'DqLoader.php';
date_default_timezone_set("PRC");
//可配置多個
$server=array(
    '127.0.0.1:6789',
);
$dqClient = new DqClient();
$dqClient->addServer($server);

$topic ='order_openvip_checker'; //topic在后臺注冊
$id = uniqid();
$data=array(
    'id'=>$id,
    'body'=>array(
        'a'=>1,
        'b'=>2,
        'c'=>3,
        'ext'=>str_repeat('a',64),
    ),
    //可選，設(shè)置后以這個通知時間為準(zhǔn)，默認(rèn)延時時間在注冊topic的時候指定
    'fix_time'=>date('Y-m-d 23:50:50'),
);

//添加
$boolRet = $dqClient->add($topic, $data);
echo 'add耗時:'.(msectime() - $time)."ms\n";
//查詢
$time = msectime();
$result = $dqClient->get($topic, $id);
echo 'get耗時:'.(msectime() - $time)."ms\n";

//刪除
$time = msectime();
$boolRet = $dqClient->del($topic,$id);
echo 'del耗時:'.(msectime() - $time)."ms\n";

執(zhí)行php test.php

默認(rèn)日志目錄在項目目錄的logs目錄下，在DqConf.php修改$logPath

• 1.請求日志：request_ymd.txt
• 2.通知日志：notify_ymd.txt
• 3.錯誤日志：err_ymd.txt

• 1.系統(tǒng)會自動檢測配置文件新，如果有改動，會自動退出(沒有找到較好的熱更新的方案)，需要重啟，可以在crontab里面建個任務(wù),1分鐘執(zhí)行一次，程序有check_self的判斷
• 2.優(yōu)雅退出命令: master檢測偵聽了USR2信號，收到信號后會通知所有子進(jìn)程，子進(jìn)程完成當(dāng)前任務(wù)后會自動退出

ps -ef | grep dq-master| grep -v grep | head -n 1 | awk '{print $2}' | xargs kill -USR2

六、性能測試

需要安裝pthreads拓展：

測試原理：使用多線程模擬并發(fā)，在1s內(nèi)能成功返回請求成功的個數(shù)

php DqBench  concurrency  requests
concurrency:并發(fā)數(shù)
requests： 每個并發(fā)產(chǎn)生的請求數(shù)

測試環(huán)境：內(nèi)存 8G ，8核cpu，2個redis和1個dq-server 部署在一個機(jī)器上，數(shù)據(jù)包64字節(jié)
qps：2400

七、值得一提的性能優(yōu)化點：

• 1.redis multi命令：將多個對redis的操作打包成一個減少網(wǎng)絡(luò)開銷
• 2.計數(shù)的操作異步處理，在異步邏輯里面用函數(shù)的static變量來保存，當(dāng)寫入redis成功后釋放static變量，可以在redis出現(xiàn)異常時計數(shù)仍能保持一致，除非進(jìn)程退出
• 3.內(nèi)存泄露檢測有必要: 所有的內(nèi)存分配在底層都是調(diào)用了brk或者mmap，只要程序只有大量brk或者mmap的系統(tǒng)調(diào)用，內(nèi)存泄露可能性非常高 ,檢測命令: strace -c -p pid | grep 'mmap| brk'
• 4.檢測程序的系統(tǒng)調(diào)用情況：strace -c -p pid ，發(fā)現(xiàn)某個系統(tǒng)函數(shù)調(diào)用是其他的數(shù)倍，可能大概率程序存在問題

八、異常處理

如果調(diào)用通知接口在超時時間內(nèi)，沒有收到回復(fù)認(rèn)為通知失敗，系統(tǒng)會重新把數(shù)據(jù)放入隊列，重新通知，系統(tǒng)默認(rèn)最大通知10次(可以在Dqconf.php文件中修改$notify_exp_nums)通知間隔為2n+1，比如第一次1分鐘，通知失敗，第二次3分鐘后，直到收到回復(fù)，超出最大通知次數(shù)后系統(tǒng)自動丟棄，同時發(fā)郵件通知

ps:網(wǎng)絡(luò)抖動在所難免，通知接口如果涉及到核心的服務(wù),一定要保證冪等?。?/code>