PHPとRedisに基づいて実装された遅延キュー-PHPチュートリアル-php.cn

これは、redis をベースとした遅延キューであり、可用性が高く、拡張が容易で、アクセスが容易で、運用環境で安定しています。遅延キューはその名のとおり、遅延機能を備えたメッセージキューです。では、どのような状況でそのようなキューが必要になるのでしょうか?

PHPとRedisに基づいて実装された遅延キュー

推奨チュートリアル: 「redis ビデオチュートリアル」

1. 背景

まずビジネスシナリオを見てみましょう:

1. メンバーシップの有効期限が切れる 3 日前にリコール通知を送信します

2. 注文の支払いが正常に完了したら、次のことを確認します5 分後にダウンストリームリンクがすべて接続されているかどうか通常、たとえば、ユーザーがメンバーシップを購入した後、すべてのメンバーシップステータスが正常に設定されているかどうか

3. 返金ステータスの注文が正常に設定されているかどうかを定期的に確認する方法返金されましたか？

4. 通知が失敗した場合、相手が応答するまで 1 分、3 分、5 分、7 分後に通知が繰り返されますか?

通常、上記の問題を解決する最も簡単かつ直接的な方法は、メーターを定期的にスキャンすることです。

テーブルスキャンの問題点は次のとおりです:

1. テーブルスキャンはデータベースに長時間接続されます。大量の場合、接続は異常な中断を起こしやすくなります。より多くの例外処理とプログラムが必要になる高い堅牢性要件

2. データ量が多いと遅延が大きくなり、規定内に処理が完了できずビジネスに影響を与える。処理を開始すると、追加のメンテナンスコストが発生し、根本的な解決にはなりません。

3. 各企業は独自のテーブルスキャンロジックを維持する必要があります。ビジネスがますます成長すると、テーブルスキャン部分のロジックが繰り返し開発されることがわかりますが、それは非常によく似ています。

#遅延キューは上記のニーズを非常にうまく解決できます

2. 調査

次のように、市場にあるいくつかのオープンソースソリューションを調査しました:

1. Youzan テクノロジー: 原則だけで、原則はありません。オープンソースコード

2. github Personal: https://github.com/ouqiang/lay-queue

1.基于redis实现，redis只能配置一个,如果redis挂了整个服务不可用，可用性差点
2.消费端实现的是拉模式，接入成本大，每个项目都得去实现一遍接入代码
3.在star使用的人数不多，放在生产环境，存在风险，加之对go语言不了解，出了问题难以维护

ログイン後にコピー

3. SchedulerX-Alibaba オープンソース: 非常に強力ですが、運用とメンテナンスが困難です。複雑で多くのコンポーネントに依存しており、軽量ではありません

4.RabbitMQ 遅延タスク: 遅延関数がないため、機能を利用して独自に実装する必要があります。会社はこのキューを導入していません。遅延キューを作るために別のものを導入するのは少し費用がかかりますし、費用もかかります。メンテナンスには特別な運用と保守が必要です。現在、チームは

# をサポートしていません##基本的には上記の理由から自分で書く予定です普段はPHPを主に使っていますプロジェクトの基本的なredis zset構造はストレージとして使用し、PHP言語で実装しています実装原理についてはLikeを参照してくださいチーム: https://tech.youzan.com/queuing_lay/

遅延キュー全体には主に 4 つの部分があります:

1.JobPool はすべてを保存するために使用されます。ジョブのメタ情報。

2.DelayBucket は時間をディメンションとする順序付きキューのセットで、遅延する必要があるすべてのジョブを保存するために使用されます (ここにはジョブ ID のみが保存されます)。

3.タイマーは、各バケットをリアルタイムでスキャンし、遅延時間が現在時刻以上であるジョブを対応する準備完了キューに配置する役割を果たします。

4.ReadyQueue は、コンシューマープログラムで使用できるように、ジョブを準備完了状態で保存します (ここには JobId のみが保存されます)。

PHPとRedisに基づいて実装された遅延キュー

メッセージ構造各ジョブには次の属性が含まれている必要があります:

1.topic: ジョブタイプ。具体的な企業名として理解できます。

2.id: ジョブの一意の識別子。指定したジョブ情報を取得、削除する場合に使用します。

3.layTime: jod 遅延実行時間、13 桁のタイムスタンプ

4.ttr (time-to-run): ジョブ実行タイムアウト。

5.body: コンシューマーが特定の業務処理を実行するためのジョブの内容。json 形式で保存されます。

同じタイプのトピック遅延時間の場合、ttr は通常固定されており、ジョブプロパティは簡素化できます。

1.topic: ジョブタイプ。これは、特定のビジネス名

2.id: ジョブの一意の識別子として理解できます。指定したジョブ情報を取得、削除する場合に使用します。

3.body: コンシューマーが特定の業務処理を実行するためのジョブの内容。json 形式で保存されます。

#遅延時間、ttr は、トピック管理者のバックグラウンドで構成されます

#3.ターゲット

Light Magnitude
: swoole、workman などのネットワークフレームワークを導入せずに、少ない PHP 拡張機能で直接実行できます
: を使用マスターワークアーキテクチャでは、マスターは業務処理を行わず、子プロセスの管理のみを担当し、子プロセスが異常終了した場合は自動的にプルアップされます
:
1. マルチインスタンスの展開をサポートし、各インスタンスはステートレスであり、1 つのインスタンスの障害はサービスに影響しません

2複数の Redis の構成をサポートし、1 つの Redis の障害はサービスの一部のメッセージにのみ影響します

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に基づいて実装された遅延キュー