首頁 > 文章 > 後端開發 > 基於php和redis實現的延遲佇列

基於php和redis實現的延遲佇列

藏色散人原創: 2019-01-04 10:22:249693瀏覽

基於redis實現高可用，易拓展，接入方便，生產環境穩定運行的延遲隊列。延遲隊列，顧名思義它是一種帶有延遲功能的訊息隊列。那麼，是在什麼場景下我才需要這樣的隊列呢？

基於php和redis實現的延遲佇列

推薦教學：《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個部分：基於php和redis實現的延遲佇列

1.JobPool用來存放所有Job的元資訊。 2.DelayBucket是一組以時間為維度的有序佇列，用來存放所有需要延遲的Job（這裡只存放Job Id）。

3.Timer負責即時掃描各個Bucket，並將delay時間大於等於目前時間的Job放入對應的Ready Queue。

4.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不做業務處理，只負責管理子程序，子程序異常退出時自動拉起

可用性

1.支援多實例部署，每個實例無狀態，一個實例掛掉不影響服務
3.業務方存取方便，在後台只需填寫相關訊息類型和回呼介面
######拓展性###: 當消費進程存在瓶頸時，可以配置加大消費進程數，當寫入存在瓶頸時，可增加實例數寫入效能可線性提高###############即時性### ：允許存在一定的時間誤差。 ###
支持消息删除：业务使用方，可以随时删除指定消息。
消息传输可靠性：消息进入到延迟队列后，保证至少被消费一次。
写入性能：qps>1000+

四、架构设计与说明

总体架构

基於php和redis實現的延遲佇列

采用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

五、模块流程图

消息写入:

基於php和redis實現的延遲佇列

timer查找到期消息:

基於php和redis實現的延遲佇列

consumer消费流程:

基於php和redis實現的延遲佇列

六、部署

环境依赖：PHP 5.5+ 安装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了路径 $phpBin

命令：

php DqHttpServer.php --port 8088

访问:http://127.0.0.1:8088,出现配置界面

基於php和redis實現的延遲佇列

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

step4:配置告信息(比如redis宕机)

基於php和redis實現的延遲佇列

step5:注册topic

基於php和redis實現的延遲佇列

重试标记说明:

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}==&#39;返回失败&#39;

基於php和redis實現的延遲佇列

step6:启动服务进程:

php DqInit.php --port 6789 &

执行 ps -ef | grep dq 看到如下信息说明启动成功

基於php和redis實現的延遲佇列

step7: 写入数据，参考demo.php

step8:查看日志

默认日志目录在项目目录的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 &#39;{print $2}&#39; | 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 &#39;mmap| brk&#39;

4.检测程序的系统调用情况：strace -c -p pid ，发现某个系统函数调用是其他的数倍，可能大概率程序存在问题

推荐参考：《Redis命令操作中文手册》

九、异常处理

1.如果调用通知接口在超时时间内，没有收到回复认为通知失败，系统会重新把数据放入队列，重新通知，系统默认最大通知10次(可以在Dqconf.php文件中修改$notify_exp_nums)通知间隔为2n+1，比如第一次1分钟，通知失败，第二次3分钟后，直到收到回复，超出最大通知次数后系统自动丢弃，同时发邮件通知

2.线上redis每隔1s持久化一次，存在丢失1s数据的情况，出现这种情况可以对比request_ymd.txt和notify_ymd.txt日志手动恢复过来

3.redis宕机通知:

基於php和redis實現的延遲佇列