纯php+socket实现延迟队列

2019独角兽企业重金招聘Python工程师标准>>>

延迟队列&＃xff0c;顾名思义它是一种带有延迟功能的消息队列。那么&＃xff0c;是在什么场景下我才需要这样的队列呢&＃xff1f;

一、背景

先看看一下业务场景&＃xff1a;

1.会员过期前3天发送召回通知
2.订单支付成功后&＃xff0c;5分钟后检测下游环节是否都正常&＃xff0c;比如用户购买会员后&＃xff0c;各种会员状态是否都设置成功
3.如何定期检查处于退款状态的订单是否已经退款成功&＃xff1f;
4.实现通知失败&＃xff0c;1&＃xff0c;3&＃xff0c;5&＃xff0c;7分钟重复通知&＃xff0c;直到对方回复&＃xff1f;

通常解决以上问题&＃xff0c;最简单直接的办法就是定时去扫表。

扫表存在的问题是&＃xff1a;

1.扫表与数据库长时间连接&＃xff0c;在数量量大的情况容易出现连接异常中断&＃xff0c;需要更多的异常处理&＃xff0c;对程序健壮性要求高
2.在数据量大的情况下延时较高&＃xff0c;规定内处理不完&＃xff0c;影响业务&＃xff0c;虽然可以启动多个进程来处理&＃xff0c;这样会带来额外的维护成本&＃xff0c;不能从根本上解决。
3.每个业务都要维护一个自己的扫表逻辑。当业务越来越多时&＃xff0c;发现扫表部分的逻辑会重复开发&＃xff0c;但是非常类似

延时队列能对于上述需求能很好的解决

二、调研

调研了市场上一些开源的方案&＃xff0c;以下&＃xff1a;

1.有赞科技&＃xff1a;只有原理&＃xff0c;没有开源代码
2.github个人的&＃xff1a;https://github.com/ouqiang/delay-queue

1.基于redis实现&＃xff0c;redis只能配置一个,如果redis挂了整个服务不可用&＃xff0c;可用性差点 2.消费端实现的是拉模式&＃xff0c;接入成本大&＃xff0c;每个项目都得去实现一遍接入代码 3.在star使用的人数不多&＃xff0c;放在生产环境&＃xff0c;存在风险&＃xff0c;加之对go语言不了解&＃xff0c;出了问题难以维护
3.SchedulerX-阿里开源的&＃xff1a; 功能很强大&＃xff0c;但是运维复杂&＃xff0c;依赖组件多&＃xff0c;不够轻量
4.RabbitMQ-延时任务: 本身没有延时功能&＃xff0c;需要借助一特性自己实现&＃xff0c;而且公司没有部署这个队列&＃xff0c;去单独部署一个这个来做延时队列成本有点高&＃xff0c;而且还需要专门的运维来维护&＃xff0c;目前团队不支持

基本以上原因打算自己写一个&＃xff0c;平常使用php多&＃xff0c;项目基本redis的zset结构作为存储&＃xff0c;用php语言实现 &＃xff0c;实现原理参考了有赞团队&＃xff1a;https://tech.youzan.com/queuing_delay/

三、目标

轻量级&＃xff1a;有较少的php的拓展就能直接运行&＃xff0c;不需要引入网络框架&＃xff0c;比如swoole&＃xff0c;workman之类的
稳定性&＃xff1a;采用master-work架构&＃xff0c;master不做业务处理&＃xff0c;只负责管理子进程&＃xff0c;子进程异常退出时自动拉起
可用性&＃xff1a;
- 1.支持多实例部署&＃xff0c;每个实例无状态&＃xff0c;一个实例挂掉不影响服务
- 2.支持配置多个redis&＃xff0c;一个redis挂了只影响部分消息
- 3.业务方接入方便&＃xff0c;在后台只需填写相关消息类型和回掉接口
拓展性: 当消费进程存在瓶颈时&＃xff0c;可以配置加大消费进程数&＃xff0c;当写入存在瓶颈时&＃xff0c;可增加实例数写入性能可线性提高
实时性&＃xff1a;允许存在一定的时间误差。
支持消息删除&＃xff1a;业务使用方&＃xff0c;可以随时删除指定消息。
消息传输可靠性&＃xff1a;消息进入到延迟队列后&＃xff0c;保证至少被消费一次。
写入性能&＃xff1a;qps>1000&＃43;

四、架构设计与说明

总体架构在这里插入图片描述

采用master-work架构模式&＃xff0c;主要包括6个模块&＃xff1a;

1.dq-mster: 主进程&＃xff0c;负责管理子进程的创建&＃xff0c;销毁&＃xff0c;回收以及信号通知
2.dq-server: 负责消息写入&＃xff0c;读取&＃xff0c;删除功能以及维护redis连接池
3.dq-timer-N: 负责从redis的zset结构中扫描到期的消息&＃xff0c;并负责写入ready 队列&＃xff0c;个数可配置&＃xff0c;一般2个就行了&＃xff0c;因为消息在zset结构是按时间有序的
4.dq-consume-N: 负责从ready队列中读取消息并通知给对应回掉接口&＃xff0c;个数可配置
5.dq-redis-checker: 负责检查redis的服务状态&＃xff0c;如果redis宕机&＃xff0c;发送告警邮件
6.dq-http-server: 提供web后台界面&＃xff0c;用于注册topic

五、部署

环境依赖&＃xff1a;PHP 5.4&＃43; 安装sockets&＃xff0c;redis&＃xff0c;pcntl,pdo_mysql 拓展

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

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

step2:在DqConfg.文件中配置数据库信息&＃xff1a; DqConf::$db

step3: 启动http服务

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

命令&＃xff1a;

php DqHttpServer.php --port 8088

访问:http://127.0.0.1:8088,出现配置界面在这里插入图片描述

redis信息格式&＃xff1a;host:post:auth 比如 127.0.0.1:6379:12345

stop4:启动服务进程:

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

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

在这里插入图片描述

stop6:注册topic

在这里插入图片描述

step7: 写入数据&＃xff0c;在项目根目录下新建test.php文件写入

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

执行php test.php

step8:查看日志

默认日志目录在项目目录的logs目录下&＃xff0c;在DqConf.php修改$logPath

1.请求日志&＃xff1a;request_ymd.txt
2.通知日志&＃xff1a;notify_ymd.txt
3.错误日志&＃xff1a;err_ymd.txt

step9:如果配置文件有改动

1.系统会自动检测配置文件新&＃xff0c;如果有改动&＃xff0c;会自动退出(没有找到较好的热更新的方案)&＃xff0c;需要重启&＃xff0c;可以在crontab里面建个任务,1分钟执行一次&＃xff0c;程序有check_self的判断
2.优雅退出命令: master检测侦听了USR2信号&＃xff0c;收到信号后会通知所有子进程&＃xff0c;子进程完成当前任务后会自动退出

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

六、性能测试

需要安装pthreads拓展&＃xff1a;

测试原理&＃xff1a;使用多线程模拟并发&＃xff0c;在1s内能成功返回请求成功的个数

php DqBench concurrency requests concurrency:并发数 requests&＃xff1a; 每个并发产生的请求数测试环境&＃xff1a;内存 8G &＃xff0c;8核cpu&＃xff0c;2个redis和1个dq-server 部署在一个机器上&＃xff0c;数据包64字节 qps&＃xff1a;2400

七、值得一提的性能优化点&＃xff1a;

1.redis multi命令&＃xff1a;将多个对redis的操作打包成一个减少网络开销
2.计数的操作异步处理&＃xff0c;在异步逻辑里面用函数的static变量来保存&＃xff0c;当写入redis成功后释放static变量&＃xff0c;可以在redis出现异常时计数仍能保持一致&＃xff0c;除非进程退出
3.内存泄露检测有必要: 所有的内存分配在底层都是调用了brk或者mmap&＃xff0c;只要程序只有大量brk或者mmap的系统调用&＃xff0c;内存泄露可能性非常高 ,检测命令: strace -c -p pid | grep &＃39;mmap| brk&＃39;
4.检测程序的系统调用情况&＃xff1a;strace -c -p pid &＃xff0c;发现某个系统函数调用是其他的数倍&＃xff0c;可能大概率程序存在问题

八、异常处理

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

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