php延遲隊列是如何實現的

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

一、背景

先看看業務場景：

1.會員過期前3天發送召回通知

# 2.訂單付款成功後，5分鐘後檢測下游環節是否都正常，例如用戶購買會員後，各種會員狀態是否都設定成功

3.如何定期檢查處於退款狀態的訂單是否已經退款成功？

4.實現通知失敗，1，3，5，7分鐘重複通知，直到對方回覆？

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

掃表存在的問題是：

1.掃表與資料庫長時間連接，在數量量大的情況容易出現連接異常中斷，需要更多的異常處理，對程序健壯性要求高

2.在資料量大的情況下延時較高，規定內處理不完，影響業務，雖然可以啟動多個進程來處理，這樣會帶來額外的維護成本，不能從        根本解決。

3.每個業務都要維護一個自己的掃表邏輯。當業務越來越多時，發現掃表部分的邏輯會重複開發，但是非常類似

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

二、研究

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

·有讚科技：只有原理，沒有開源程式碼

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

    1.基於redis實現，redis只能配置一個,如果redis掛了整個服務不可用，可用性差點

    2.消費端實現的是拉模式，接入成本大，每個專案都得去實現一次接入代碼

    3.在star使用的人數不多，放在生產環境，有風險，加上對go語言不了解，出了問題難以維護

·SchedulerX-阿里開源的： 功能很強大，但是維運複雜，依賴元件多，不夠輕量

·RabbitMQ-延時任務: 本身沒有延時功能，需要藉助一特性自己實現，而且公司沒有部署這個隊列，去單獨部署一個這個來做延時佇列成本有點高，而且還需要專門的運維來維護，目前團隊不支援

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

相關推薦：《php教程》

三、目標

·輕量級：有較少的php的拓展就能直接運行，不需要引入網路框架，例如swoole，workman之類的

·穩定性：採用master-work架構，master不做業務處理，只負責管理子進程，子程序異常退出時自動拉起

#·可用性：

    1.支援多重執行個體部署，每個執行個體無狀態，一個執行個體掛掉不影響服務

    2.支援設定多個redis，一個redis掛了只影響部分訊息

    3.業務方存取方便，在後台只需填寫相關訊息類型和回掉介面

·拓展性: 當消費進程存在瓶頸時，可以配置加大消費進程數，當寫入存在瓶頸時，可增加實例數寫入效能可線性提高

·實時性：允許存在一定的時間誤差。

·支援訊息刪除：業務使用方，可以隨時刪除指定訊息。

·訊息傳輸可靠性：訊息進入延遲佇列後，保證至少被消費一次。

·寫入效能：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##########五、部署########環境依賴：PHP 5.4 安裝sockets，redis，pcntl,pdo_mysql 拓展###

step1：安装数据库用于存储一些topic以及告警信息

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;

step2：在DqConfg.文件中配置数据库信息： DqConf::$db

step3：启动http服务

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

命令：

php DqHttpServer.php --port 8088

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

在这里插入图片描述

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

stop4：启动服务进程

php DqInit.php --port 6789

看到如下信息说明启动成功

在这里插入图片描述

stop5：配置告警信息(比如redis宕机)

在这里插入图片描述

stop6：注册topic

在这里插入图片描述

在这里插入图片描述

step7：写入数据，在项目根目录下新建test.php文件写入

<?php
include_once &#39;DqLoader.php&#39;;
date_default_timezone_set("PRC");
//可配置多个
$server=array(
    &#39;127.0.0.1:6789&#39;,
);
$dqClient = new DqClient();
$dqClient->addServer($server);
 
$topic =&#39;order_openvip_checker&#39;; //topic在后台注册
$id = uniqid();
$data=array(
    &#39;id&#39;=>$id,
    &#39;body&#39;=>array(
        &#39;a&#39;=>1,
        &#39;b&#39;=>2,
        &#39;c&#39;=>3,
        &#39;ext&#39;=>str_repeat(&#39;a&#39;,64),
    ),
    //可选，设置后以这个通知时间为准，默认延时时间在注册topic的时候指定
    &#39;fix_time&#39;=>date(&#39;Y-m-d 23:50:50&#39;),
);
 
//添加
$boolRet = $dqClient->add($topic, $data);
echo &#39;add耗时:&#39;.(msectime() - $time)."ms\n";
//查询
$time = msectime();
$result = $dqClient->get($topic, $id);
echo &#39;get耗时:&#39;.(msectime() - $time)."ms\n";
 
//删除
$time = msectime();
$boolRet = $dqClient->del($topic,$id);
echo &#39;del耗时:&#39;.(msectime() - $time)."ms\n";

执行php test.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 '{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 'mmap| brk'。

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

八、异常处理

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

ps:网络抖动在所难免，通知接口如果涉及到核心的服务,一定要保证幂等！！

九、线上情况

线上部署了两个实例每个机房部一个，4个redis作存储，服务稳定运行数月，各项指标均符合预期。

主要接入业务：

    ·订单10分钟召回通知

    ·接口超时或者失败补偿

项目地址: https://github.com/chenlinzhong/php-delayqueue

以上是php延遲隊列是如何實現的的詳細內容。更多資訊請關注PHP中文網其他相關文章！