订单15分钟未支付超时取消延时队列

调研了市场上一些开源的方案，以下：

三、目标

• 轻量级：有较少的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 拓展

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

stop4:启动服务进程:

php DqInit.php --port 6789
看到如下信息说明启动成功

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),
    ),
    //可选，设置后以这个通知时间为准，默认延时时间在注册topic的时候指定
    'fix_time'=>date('Y-m-d 23:50:50'),
);

//添加
$boolRet = $dqClient->add($topic, $data);
echo 'add耗时:'.(msectime() - $time)."msn";
//查询
$time = msectime();
$result = $dqClient->get($topic, $id);
echo 'get耗时:'.(msectime() - $time)."msn";

//删除
$time = msectime();
$boolRet = $dqClient->del($topic,$id);
echo 'del耗时:'.(msectime() - $time)."msn";

执行php test.php

默认日志目录在项目目录的logs目录下，在DqConf.php修改$logPath

• 1.请求日志：request_ymd.txt
• 2.通知日志：notify_ymd.txt
• 3.错误日志：err_ymd.txt

• 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分钟召回通知
• 接口超时或者失败补偿