PHP基于Redis实现轻量级延迟队列

时间 2019-11-29

标签 php 基于 redis 实现轻量级延迟队列栏目 PHP 繁體版

原文原文链接

延迟队列，顾名思义它是一种带有延迟功能的消息队列。那么，是在什么场景下我才须要这样的队列呢？php

1、背景

先看看一下业务场景：html

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

一般解决以上问题，最简单直接的办法就是定时去扫表。mysql

扫表存在的问题是：git

1.扫表与数据库长时间链接，在数量量大的状况容易出现链接异常中断，须要更多的异常处理，对程序健壮性要求高
2.在数据量大的状况下延时较高，规定内处理不完，影响业务，虽然能够启动多个进程来处理，这样会带来额外的维护成本，不能从根本上解决。
3.每一个业务都要维护一个本身的扫表逻辑。当业务愈来愈多时，发现扫表部分的逻辑会重复开发，可是很是相似

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

2、调研

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

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/redis

整个延迟队列主要由4个部分sql

JobPool用来存放全部Job的元信息。
DelayBucket是一组以时间为维度的有序队列，用来存放全部须要延迟的Job（这里只存放Job Id）。
Timer负责实时扫描各个Bucket，并将delay时间大于等于当前时间的Job放入到对应的Ready Queue。
ReadyQueue存放处于Ready状态的Job（这里只存放JobId），以供消费程序消费。
- - 　
    
    消息结构
    每一个Job必须包含一下几个属性：docker
    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后台配置
    
    3、目标
    - 轻量级：有较少的php的拓展就能直接运行，不须要引入网络框架，好比swoole，workman之类的
    - 稳定性：采用master-work架构，master不作业务处理，只负责管理子进程，子进程异常退出时自动拉起
    - 可用性：
    - 拓展性: 当消费进程存在瓶颈时，能够配置加大消费进程数，当写入存在瓶颈时，可增长实例数写入性能可线性提升
    - 实时性：容许存在必定的时间偏差。
    - 支持消息删除：业务使用方，能够随时删除指定消息。
    - 消息传输可靠性：消息进入到延迟队列后，保证至少被消费一次。
    - 写入性能：qps>1000+
    4、架构设计与说明
    
    整体架构
  - 采用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
    5、模块流程图
    
    消息写入:
  - timer查找到期消息:
  - timer查找到期消息:
    
    consumer消费流程:
    
    6、部署
    
    环境依赖：PHP 5.4+ 安装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了路径
    
    命令：
    
    php DqHttpServer.php --port 8088
    
    访问:http://127.0.0.1:8088,出现配置界面
    
    redis信息格式：host:port:auth 好比 127.0.0.1:6379:12345
    
    stop4:配置告信息(好比redis宕机)
    
    stop5:注册topic
    
    重试标记说明:
    
    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}== '返回失败'
    
    step6:启动服务进程:
    
    php DqInit.php --port 6789 &
    
    执行 ps -ef | grep dq 看到以下信息说明启动成功
    
    step7: 写入数据，参考demo.phpstep8:查看日志
    
    默认日志目录在项目目录的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
    
    7、性能测试
    须要安装pthreads拓展：
    
    测试原理：使用多线程模拟并发，在1s内能成功返回请求成功的个数
    
    php DqBench concurrency requests
    
    concurrency:并发数
    
    requests：每一个并发产生的请求数
    
    测试环境：内存 8G ，8核cpu，2个redis和1个dq-server 部署在一个机器上，数据包64字节
    
    qps：2400
    
    8、值得一提的性能优化点：
    - 1.redis multi命令：将多个对redis的操做打包成一个减小网络开销
    - 2.计数的操做异步处理，在异步逻辑里面用函数的static变量来保存，当写入redis成功后释放static变量，能够在redis出现异常时计数仍能保持一致，除非进程退出
    - 3.内存泄露检测有必要: 全部的内存分配在底层都是调用了brk或者mmap，只要程序只有大量brk或者mmap的系统调用，内存泄露可能性很是高 ,检测命令: strace -c -p pid | grep -P 'mmap| brk'
    - 4.检测程序的系统调用状况：strace -c -p pid ，发现某个系统函数调用是其余的数倍，可能大几率程序存在问题
    9、异常处理
    1.若是调用通知接口在超时时间内，没有收到回复认为通知失败，系统会从新把数据放入队列，从新通知，系统默认最大通知10次(能够在Dqconf.php文件中修改$notify_exp_nums)通知间隔为2n+1，好比第一次1分钟，通知失败，第二次3分钟后，直到收到回复，超出最大通知次数后系统自动丢弃，同时发邮件通知
    
    2.线上redis每隔1s持久化一次，存在丢失1s数据的状况，出现这种状况能够对比request_ymd.txt和notify_ymd.txt日志手动恢复过来
    
    3.redis宕机通知:
    
    ps:网络抖动在所不免，通知接口若是涉及到核心的服务,必定要保证幂等！！
    
    10、线上状况
    
    线上部署了两个实例每一个机房部一个，4个redis共16G内存做存储，服务稳定运行数月，各项指标均符合预期
    
    主要接入业务:
    - 订单10分钟召回通知
    - 调用接口超时或者失败时作补偿
    - 会员过时前3天召回通知
    11、不足与展望
    1.因为团队使用的镜像缺乏libevent拓展，因此dq-server基于select模型，并发高的场景下性能存在瓶颈，后续能够改成基于libevent事件模型，提高并发性能
    
    2.timer和consumer目前是采用多进程来作的，这个粒度感受有点粗，能够考虑使用多线程模式，而且支持动态建立线程数来提升consumer的性能，最大程度保证消费及时
    
    3.dq-server与redis是同步调用，这也是性能的瓶颈点，计划基于swoole_redis来异步处理
  - 推荐阅读：
  - Swoole和Redis实现的并发队列处理系统

PHP基于Redis实现轻量级延迟队列

1、背景

2、调研

3、目标

4、架构设计与说明

5、模块流程图