PHP は、Redis に基づいて軽量の遅延キュー (マルチスレッド) を実装します。

遅延キューは、その名のとおり、遅延機能を備えたメッセージキューです。では、どのような状況でそのようなキューが必要になるのでしょうか?

1. 背景

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

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

2. 注文の支払いが完了したら、5 分後にダウンストリーム リンクが正常であるかどうかを確認します (たとえば、ユーザーがメンバーシップを購入した後、さまざまなメンバーシップ ステータスが正常に設定されているかどうかなど)。返金ステータスの注文が正常に返金されたかどうかを定期的に確認するにはどうすればよいですか?

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

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

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

1. テーブル スキャンではデータベースへの接続時間が長く、接続数が多いと接続が異常中断されやすくなります。例外処理とプログラムの詳細 高い堅牢性要件

2. データ量が多いと遅延が大きく、規定内に処理が完了できずビジネスに影響を与える 複数のプロセスを起動できるが、加工を行うとメンテナンス費用が発生し、根本的な解決にはなりません。

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

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

2. 調査

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

1. Youzan Technology: 原則のみ、オープン ソース コードなし

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

(1) Redis の実装に基づいて、構成できる Redis は 1 つだけです。Redis がハングすると、サービス全体が利用できなくなります。可用性はほぼ

( 2) コンシューマ側はプル モデルを実装しており、アクセス コストが高くなります。各プロジェクトでアクセス コードを実装する必要があります

(3) 人が多くありませんスターで使用するため、本番環境に置くにはリスクがあり、Go 言語の理解が不足しているため、問題が発生した場合の保守が困難です。

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

4 .RabbitMQ 遅延タスク: それ自体には遅延関数がありません。さらに、同社はこのキューをデプロイしていません。遅延キューを作成するために個別にデプロイするには少し費用がかかり、また、それを維持するために特別な運用とメンテナンスも必要です。現在、チームはデプロイしています。

基本的には上記の理由から自分で書くつもりです 普段はPHPを使っています このプロジェクトは基本的にPHP言語で実装されているredisのzset構造をストレージとして利用しています 実装原理についてYouzan チームを参照してください: https://tech.youzan.com/queuing_delay/

遅延キュー全体は主に 4 つの部分で構成されています

JobPool はメタ情報を保存するために使用されますすべての仕事の。

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

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

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

メッセージ構造

各ジョブには次の属性が含まれている必要があります:

トピック: ジョブ タイプ。具体的な企業名として理解できます。

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

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

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

body: 消費者が特定の業務処理を実行するためのジョブ コンテンツ。json 形式で保存されます。

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

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

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

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

遅延時間、ttr は topicadmin 背景で設定されます

3. ターゲット

軽量: PHP の拡張を少なくして直接実行できます。

安定性: マスターワークアーキテクチャを使用すると、マスターはビジネス処理を実行せず、子プロセスの管理のみを担当します。子プロセスが異常終了しました。自動的に開始されます。

可用性:

1. マルチインスタンスのデプロイメントをサポートし、各インスタンスはステートレスであり、1 つのインスタンスの障害はサービスに影響しません

2. 1 つの Redis に障害が発生した場合でも、複数の Redis の構成をサポートします 一部のメッセージのみに影響します

##3. ビジネス側は簡単にアクセスでき、関連するメッセージ タイプとコールバック インターフェイスを入力するだけで済みますバックグラウンドで

拡張性: 消費プロセスにボトルネックがある場合、消費プロセスの数を増やすように構成できます。書き込みにボトルネックがある場合、インスタンスの数を増やすことができます。

リアルタイム パフォーマンス: 一定の時間誤差は許容されます。

サポート メッセージ削除: ビジネス ユーザーは、指定されたメッセージをいつでも削除できます。

メッセージ送信の信頼性: メッセージが遅延キューに入った後、少なくとも 1 回は消費されることが保証されます。

書き込みパフォーマンス: 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

五、模块流程图

消息写入:

timer查找到期消息:

consumer消费流程:

六、部署

环境依赖：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.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内能成功返回请求成功的个数

八、值得一提的性能优化点：

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 ，发现某个系统函数调用是其他的数倍，可能大概率程序存在问题

九、异常处理

1. タイムアウト期間内に通知インターフェイスが呼び出され、応答が受信されない場合、通知は失敗したとみなされます。システムはデータを再度キューに入れ、再度通知します。システムのデフォルトは最大通知数 10 です。回 (Dqconf.php ファイル $notify_exp_nums で変更できます) 通知間隔は 2n 1 です。たとえば、初めて通知が 1 分間失敗した場合、2 回目は 3 分後に応答を受信するまで、最大通知数を超えるとシステムが自動的に破棄し、同時にメール通知を送信します

2 .オンライン Redis は 1 秒ごとに保持されます。1 秒のデータが失われる場合があります。この場合

3.redis ダウンタイム通知:

ps: ネットワーク ジッターは次のとおりです。通知インターフェイスがコア サービスに関係する場合、冪等である必要があります。 ！

10. オンラインの状況

2 つのインスタンスがオンラインでデプロイされ、各コンピューター室に 1 つずつ、合計 16G のメモリを備えた 4 つの Redis がストレージとして使用され、サービスは数か月間安定して稼働しており、すべての指標が期待どおりです。

メイン アクセス ビジネス:

##·10 分間のリコール通知を注文する

· インターフェースの呼び出しがタイムアウトまたは失敗した場合の補償

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

11. 欠点チームが使用するイメージには libevent 拡張機能がないため、dq-server は選択モデルに基づいており、同時実行性の高いシナリオではパフォーマンスのボトルネックが発生します。同時実行パフォーマンスを向上させるために、将来的には libevent イベント モデルに基づく予定です。 2. タイマーとコンシューマは現在、複数のプロセスを使用して実装されています。この粒度は少し荒いように感じます。マルチスレッド モードの使用を検討し、スレッド数の動的作成をサポートして、コンシューマのパフォーマンスを向上させ、タイムリーな消費を確保することができます。最大限の範囲です。

3.dq-serverとredisが同期的に呼ばれるのが性能のボトルネックでもありますが、swoole_redisをベースに非同期で処理する予定です。

PHP 中国語 Web サイトには、無料の

PHP ビデオ チュートリアル

が多数あり、誰でも学習することができます。

この記事はhttps://www.jianshu.com/p/58f10ac42162から転載されました。