Comment la file d'attente PHP est-elle implémentée ?

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 :

·Technologie Youzan : Que des principes, pas de code open source

·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. La convivialité est médiocre

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 gens qui utilisent star. Il y a des risques lorsqu'on est placé dans un environnement de production. De plus, si vous ne comprenez pas le langage Go, il sera difficile de le maintenir en cas de problème

. ·SchedulerX - L'open source d'Alibaba : fonctionnement et maintenance très puissants, mais complexes, dépendants de composants Trop nombreux, pas assez légers

·Tâche retardée RabbitMQ : il n'a pas une fonction de retard elle-même, elle doit être implémentée par elle-même à l'aide d'une fonctionnalité, et l'entreprise n'a pas déployé cette file d'attente, donc déployez-la séparément. Le coût de création d'une file d'attente retardée est un peu élevé et nécessite également des mesures spéciales. fonctionnement et maintenance. 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 généralement PHP principalement, et le projet utilise essentiellement la structure zset de redis comme stockage. Implémenté en langage PHP, le principe d'implémentation fait référence à l'équipe Youzan : https://tech.youzan.com/queuing_delay/

Recommandations associées : "tutoriel php"

3. Objectif

·Léger : il peut fonctionner directement avec moins d'extensions PHP, sans avoir besoin d'introduire des frameworks réseau, tels que swoole, workman, etc.

·Stabilité : grâce à l'architecture maître-travail, le maître n'effectue pas de traitement métier, mais est uniquement responsable de la gestion du processus enfant. Lorsque le processus enfant se termine anormalement, il démarre automatiquement. up

·Disponibilité :

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

2. Prend en charge la configuration de plusieurs Redis, un Redis Raccrocher n'affectera que certains messages

3. Le côté commercial a un accès facile et n'a qu'à remplir le type de message approprié et l'interface de rappel dans le contexte

·Extensibilité : lors de la consommation Lorsqu'il y a un goulot d'étranglement dans le processus, vous pouvez configurer pour augmenter le nombre de processus consommateurs. Lorsqu'il y a un goulot d'étranglement dans l'écriture, vous pouvez augmenter le nombre. nombre d'instances. Les performances d'écriture peuvent être améliorées de manière linéaire

·En temps réel : un certain niveau d'erreur de temps est autorisé.

·Prend en charge la suppression des messages : 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

Insérer la description de l'image ici

Adopte le modèle d'architecture master-work, qui comprend principalement 6 modules :

1.dq-mster : processus principal, responsable de la gestion de la création et de la destruction des enfants processus, recyclage et notification de signal

2.dq-server : responsable des fonctions d'écriture, de lecture, de suppression des messages et de la maintenance du pool de connexions redis

3.dq-timer-N : responsable de redis La structure zset analyse les messages expirés et est responsable de l'écriture dans la file d'attente prête. Le nombre est configurable, généralement 2 suffit, car les messages dans la structure zset sont classés par temps

4.dq-consume-. N : Responsable de la lecture des messages de la file d'attente prête et de la notification à l'interface de secours correspondante. Le numéro est configurable

5.dq-redis-checker : Responsable de la vérification de l'état du service de redis. Si redis est en panne, envoyez. une alarme. Email

6.dq-http-server : fournit une interface d'arrière-plan Web pour l'enregistrement du sujet

5. Déploiement

Dépendance de l'environnement : PHP. 5.4 + Installer les sockets, redis, pcntl, extension pdo_mysql

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

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

step2：在DqConfg.文件中配置数据库信息： DqConf::$db

step3：启动http服务

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

命令：

php DqHttpServer.php --port 8088

访问:http://127.0.0.1:8088,出现配置界面

在这里插入图片描述

redis信息格式：host:post:auth 比如 127.0.0.1:6379:12345

stop4：启动服务进程

php DqInit.php --port 6789

看到如下信息说明启动成功

在这里插入图片描述

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

在这里插入图片描述

stop6：注册topic

在这里插入图片描述

在这里插入图片描述

step7：写入数据，在项目根目录下新建test.php文件写入

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

执行php test.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内能成功返回请求成功的个数

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 'mmap| brk'。

4.检测程序的系统调用情况：strace -c -p pid ，发现某个系统函数调用是其他的数倍，可能大概率程序存在问题。

八、异常处理

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

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

九、线上情况

线上部署了两个实例每个机房部一个，4个redis作存储，服务稳定运行数月，各项指标均符合预期。

主要接入业务：

    ·订单10分钟召回通知

    ·接口超时或者失败补偿

项目地址: https://github.com/chenlinzhong/php-delayqueue

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!