PHP implémente une file d'attente de retard légère (multi-threading) basée sur Redis

La file d'attente de retard, comme son nom l'indique, est une file d'attente de messages avec fonction de retard. Alors, dans quelles circonstances ai-je besoin d’une telle file d’attente ?

1. Contexte

Jetons d'abord un coup d'œil au scénario commercial :

1. Envoyer une notification de rappel 3 jours avant l'expiration de l'adhésion

.

2. Une fois le paiement de la commande réussi, vérifiez si les liens en aval sont normaux après 5 minutes. Par exemple, après que l'utilisateur a acheté un abonnement, si les différents statuts d'adhésion sont définis avec succès

3. Comment. vérifier régulièrement si la commande en statut de remboursement a été remboursée avec succès ?

4. Échec de la mise en œuvre de la notification, répétez la notification dans 1, 3, 5, 7 minutes jusqu'à ce que l'autre partie réponde ?

Habituellement, le moyen le plus simple et le plus direct de résoudre les problèmes ci-dessus est de scanner régulièrement le compteur.

Les problèmes liés à l'analyse des tables sont :

1. L'analyse des tables est connectée à la base de données pendant une longue période. Dans le cas de grandes quantités, la connexion est sujette à des interruptions anormales, ce qui entraîne des interruptions anormales. nécessite plus de gestion des exceptions et modifie le programme. Exigences de robustesse élevées

2 Lorsque la quantité de données est importante, le délai est élevé et le traitement ne peut pas être effectué dans le cadre de la réglementation, ce qui affecte l'entreprise bien que plusieurs processus. peut être démarré pour le traitement, cela entraînera des coûts de maintenance supplémentaires et ne peut pas être fondamentalement résolu.

3. Chaque entreprise doit maintenir sa propre logique de numérisation des compteurs. Lorsque l'activité augmente, je trouve que la logique de la partie numérisation de table sera développée à plusieurs reprises, mais elle est très similaire

La file d'attente retardée peut très bien résoudre les besoins ci-dessus

2. Recherche

Recherche de quelques solutions open source sur le marché, comme suit :

1. Technologie Youzan : uniquement des principes, pas de code open source

2. github personnel : https://github.com/ouqiang/delay-queue

(1) Sur la base de l'implémentation de Redis, un seul Redis peut être configuré. Si Redis se bloque, l'ensemble du service sera indisponible et le service sera indisponible. la disponibilité sera presque

(2) Le côté consommateur implémente le modèle pull, et le coût d'accès est élevé. Chaque projet doit implémenter le code d'accès

(3) Il n'y a pas beaucoup de monde. l'utiliser en star, et il est risqué de le mettre dans l'environnement de production, couplé au manque de compréhension du langage Go, il est difficile à maintenir en cas de problème

3 de SchedulerX-Alibaba. open source : très puissant, mais complexe en fonctionnement et en maintenance, avec de nombreux composants dépendants, et pas assez léger

4 .Tâche retardée RabbitMQ : elle n'a pas de fonction de retard elle-même. Elle doit être implémentée par. lui-même à l'aide d'une fonctionnalité. De plus, l'entreprise n'a pas déployé cette file d'attente. Il est un peu coûteux d'en déployer une séparément pour créer une file d'attente différée, et cela nécessite également une opération et une maintenance spéciales pour la maintenir, ce que fait actuellement l'équipe. ne prend pas en charge

Fondamentalement, pour les raisons ci-dessus, je prévois d'en écrire un moi-même. J'utilise habituellement PHP. Le projet utilise essentiellement la structure zset de redis comme stockage, qui est implémentée en langage PHP pour le principe d'implémentation. , veuillez vous référer à l'équipe Youzan : https:// tech.youzan.com/queuing_delay/

L'ensemble de la file d'attente de retard est principalement composée de 4 parties

JobPool est utilisé pour stocker les méta-informations de tous les emplois.

DelayBucket est un ensemble de files d'attente ordonnées avec le temps comme dimension, utilisé pour stocker toutes les tâches qui doivent être retardées (seuls les ID de tâche sont stockés ici).

Timer est chargé d'analyser chaque bucket en temps réel et de placer les tâches dont le temps de retard est supérieur ou égal à l'heure actuelle dans la file d'attente prête correspondante.

ReadyQueue stocke les tâches à l'état Prêt (seuls les JobIds sont stockés ici) pour être consommés par les programmes grand public.

Structure du message

Chaque Job doit contenir les attributs suivants :

sujet : Type de Job. Il peut être compris comme un nom commercial spécifique.

id : L'identifiant unique du Job. Utilisé pour récupérer et supprimer les informations de tâche spécifiées.

delayTime : temps d'exécution retardé par jod, horodatage à 13 chiffres

ttr (time-to-run) : délai d'expiration de la tâche.

body : le contenu du Job, permettant aux consommateurs d'effectuer un traitement commercial spécifique, stocké au format json.

Pour le même type de délai de sujet, le ttr est généralement fixe, et les attributs du travail peuvent être simplifiés

1 sujet : Type de travail. Il peut être compris comme un nom d'entreprise spécifique

2.id : l'identifiant unique du Job. Utilisé pour récupérer et supprimer les informations de tâche spécifiées.

3.body : Le contenu du Job, permettant aux consommateurs d'effectuer un traitement commercial spécifique, stocké au format json.

delaytime, ttr sont configurés en arrière-plan du topicadmin

3. Objectif

Léger : Il peut s'exécuter directement avec moins d'extensions PHP. Il n'y en a pas. besoin d'introduire des cadres de réseau, tels que swoole, workman et autres

Stabilité : en utilisant l'architecture master-work, le maître n'effectue pas de traitement métier, il est uniquement responsable de la gestion du processus enfant, et il le fera être automatiquement extrait lorsque le processus enfant se termine anormalement

Disponibilité :

1 Prend en charge le déploiement multi-instance, chaque instance est sans état et l'échec d'une instance n'affectera pas le service

2. Prend en charge la configuration de plusieurs Redis, si un Redis échoue N'affecte que certains messages

3 Le côté commercial a un accès facile et n'a besoin que de renseigner les types de messages et les interfaces de rappel pertinents. l'arrière-plan

Évolutivité : lorsqu'il y a un goulot d'étranglement dans le processus de consommation, vous pouvez le configurer pour augmenter la consommation. Le nombre de processus. Lorsqu'il y a un goulot d'étranglement à l'écriture, le nombre d'instances peut être augmenté et l'écriture. les performances peuvent être améliorées de manière linéaire

Temps réel : une certaine erreur de temps est autorisée.

Suppression des messages de prise en charge : les utilisateurs professionnels peuvent supprimer les messages spécifiés à tout moment.

Fiabilité de la transmission des messages : une fois qu'un message entre dans la file d'attente, il est garanti qu'il sera consommé au moins une fois.

Performances d'écriture : qps>1000+

4. Conception et description de l'architecture

Architecture globale

采用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. Si l'interface de notification est appelée dans le délai d'attente et qu'aucune réponse n'est reçue, la notification est considérée comme ayant échoué. Le système mettra à nouveau les données dans la file d'attente et notifiera à nouveau par défaut une notification maximale de 10. fois (peut être modifié dans le fichier Dqconf.php $ notify_exp_nums) L'intervalle de notification est de 2n+1 Par exemple, si la notification échoue pendant 1 minute pour la première fois, jusqu'à ce qu'une réponse soit reçue après 3 minutes pour la deuxième fois, le système le supprimera automatiquement après avoir dépassé le nombre maximum de notifications et enverra une notification par e-mail en même temps

2. Redis en ligne est conservé toutes les 1 seconde, et il peut y avoir des cas où les données sont perdues pendant 1 Deuxièmement, dans ce cas, vous pouvez le restaurer manuellement en comparant les journaux request_ymd.txt et notify_ymd.txt

3. Notification de temps d'arrêt Redis :

ps. : La gigue du réseau est inévitable si l'interface de notification implique des services de base, elle doit être idempotente ! !

10. Situation en ligne

Deux instances ont été déployées en ligne, une dans chaque salle informatique, 4 redis avec un total de 16 Go de mémoire pour le stockage, et le service a été stable depuis plusieurs mois, tous les indicateurs sont conformes aux attentes.

Activité d'accès principale :

·Avis de rappel de commande de 10 minutes

·Compensation lors de l'appel de l'interface expire ou échoue

·Avis de rappel 3 jours avant l'expiration de l'adhésion

11. Lacunes et perspectives

1. Étant donné que l'image utilisée par l'équipe ne dispose pas de l'extension libevent, dq-server est basé sur le modèle select. Il existe un goulot d'étranglement en termes de performances dans les scénarios à haute concurrence. être basé sur le modèle d'événement libevent à l'avenir pour améliorer les performances de concurrence.

2. Le minuteur et le consommateur sont actuellement implémentés à l'aide de plusieurs processus. Cette granularité semble un peu approximative. Vous pouvez envisager d'utiliser le mode multithread et prendre en charge la création dynamique du nombre de threads pour améliorer les performances du consommateur et garantir une consommation rapide. la plus grande mesure.

3.dq-server et redis sont appelés de manière synchrone, ce qui constitue également un goulot d'étranglement en termes de performances. Nous prévoyons de le traiter de manière asynchrone sur la base de swoole_redis.

Site Web PHP chinois, il existe de nombreux tutoriels vidéo PHP gratuits, tout le monde est invité à apprendre !

Cet article est reproduit à partir de : https://www.jianshu.com/p/58f10ac42162