Wie wird die PHP-Verzögerungswarteschlange implementiert?

Verzögerungswarteschlange ist, wie der Name schon sagt, eine Nachrichtenwarteschlange mit Verzögerungsfunktion. Unter welchen Umständen benötige ich eine solche Warteschlange?

1. Hintergrund

Sehen wir uns zunächst das Geschäftsszenario an:

1. Rückrufbenachrichtigung 3 Tage vor Ablauf der Mitgliedschaft senden

2. Überprüfen Sie nach erfolgreicher Bestellung, ob die Downstream-Links nach 5 Minuten normal sind. Beispielsweise, nachdem der Benutzer eine Mitgliedschaft erworben hat, ob die verschiedenen Mitgliedschaftsstatus erfolgreich festgelegt wurden

3. Wie um regelmäßig zu prüfen, ob die Bestellung im Rückerstattungsstatus „Rückerstattung erfolgreich“ war?

4. Die Benachrichtigung konnte nicht implementiert werden. Wiederholen Sie die Benachrichtigung in 1, 3, 5, 7 Minuten, bis die andere Partei antwortet.

Normalerweise besteht der einfachste und direkteste Weg, die oben genannten Probleme zu lösen, darin, den Zähler regelmäßig zu scannen.

Die Probleme beim Tabellen-Scannen sind:

1. Das Tabellen-Scannen ist für eine lange Zeit mit der Datenbank verbunden. Bei großen Mengen ist die Verbindung anfällig für ungewöhnliche Unterbrechungen erfordert mehr Ausnahmebehandlung und ändert das Programm.

2 Wenn die Datenmenge groß ist, kann die Verarbeitung nicht innerhalb der Vorschriften abgeschlossen werden, was sich jedoch auf das Geschäft auswirkt kann mit der Bearbeitung begonnen werden, was zusätzliche Wartungskosten mit sich bringt, kann grundsätzlich nicht gelöst werden.

3. Jedes Unternehmen muss seine eigene Zähler-Scan-Logik pflegen. Wenn das Geschäft zunimmt, stelle ich fest, dass die Logik des Tabellenscanteils wiederholt weiterentwickelt wird, sie ist jedoch sehr ähnlich

Die Verzögerungswarteschlange kann die oben genannten Anforderungen sehr gut lösen

2. Recherche

Recherche einiger Open-Source-Lösungen auf dem Markt, wie folgt:

·Youzan-Technologie: Nur Prinzipien, kein Open-Source-Code

·Github persönlich: https://github.com/ouqiang/delay-queue

1. Basierend auf der Redis-Implementierung kann nur ein Redis konfiguriert werden, wenn Redis hängt. Der gesamte Dienst ist nicht verfügbar.

2. Die Verbraucherseite implementiert das Pull-Modell und die Zugriffskosten sind hoch. Jedes Projekt muss den Zugriffscode implementieren.

Es gibt nicht viele Leute, die Star verwenden. Wenn Sie die Go-Sprache nicht verstehen, ist es außerdem schwierig, sie zu warten

·SchedulerX – Alibabas Open Source: sehr leistungsfähig, aber komplexer Betrieb und Wartung, abhängig von Komponenten. Zu viele, nicht leicht genug

·RabbitMQ-verzögerte Aufgabe: Hat es nicht Da es sich um eine Verzögerungsfunktion selbst handelt, muss diese mithilfe einer Funktion selbst implementiert werden. Da das Unternehmen diese Warteschlange nicht bereitgestellt hat, ist die Bereitstellung dieser Funktion etwas hoch Der spezielle Betrieb und die Wartung werden vom Team derzeit nicht unterstützt.

Grundsätzlich habe ich vor, selbst eines zu schreiben. Normalerweise verwende ich PHP und das Projekt verwendet im Wesentlichen die Zset-Struktur von Redis . In PHP-Sprache implementiert, bezieht sich das Implementierungsprinzip auf das Youzan-Team: https://tech.youzan.com/queuing_delay/

Verwandte Empfehlungen: „php-Tutorial“

3. Ziel

·Leichtgewicht: Es kann direkt mit weniger PHP-Erweiterungen ausgeführt werden, ohne dass Netzwerk-Frameworks wie Swoole, Workman usw. eingeführt werden müssen .

·Stabilität: Bei Verwendung der Master-Work-Architektur führt der Master keine Geschäftsverarbeitung durch, sondern ist nur für die Verwaltung des untergeordneten Prozesses verantwortlich starten

·Verfügbarkeit:

1. Unterstützt die Bereitstellung mehrerer Instanzen, jede Instanz ist zustandslos und der Ausfall einer Instanz hat keine Auswirkungen auf den Dienst

2. Unterstützt die Konfiguration mehrerer Redis, ein Redis. Das Auflegen wirkt sich nur auf einige Nachrichten aus.

3. Die Geschäftsseite hat einfachen Zugriff und muss nur den relevanten Nachrichtentyp und die Rückrufschnittstelle eingeben der Hintergrund

·Erweiterbarkeit: Wenn es einen Engpass im Prozess gibt, können Sie konfigurieren, dass die Anzahl der konsumierenden Prozesse erhöht wird Die Anzahl der Instanzen kann linear verbessert werden

·Echtzeit: Ein bestimmter Grad ist zulässiger Zeitfehler.

·Unterstützung beim Löschen von Nachrichten: Geschäftsbenutzer können bestimmte Nachrichten jederzeit löschen.

·Zuverlässigkeit der Nachrichtenübertragung: Nachdem eine Nachricht in die Verzögerungswarteschlange gelangt ist, wird sie garantiert mindestens einmal verbraucht.

·Schreibleistung: qps>1000+

4. Architekturdesign und -beschreibung

Gesamtarchitektur

Bildbeschreibung hier einfügen

Übernimmt das Master-Work-Architekturmodell, das hauptsächlich 6 Module umfasst:

1.dq-mster: Hauptprozess, verantwortlich für die Verwaltung der Erstellung und Zerstörung von Kindern Prozesse, Recycling und Signalbenachrichtigung

2.dq-server: Verantwortlich für das Schreiben, Lesen, Löschen von Nachrichten und die Pflege des Redis-Verbindungspools

3.dq-timer-N: Verantwortlich für Redis Die zset-Struktur scannt die abgelaufenen Nachrichten und ist für das Schreiben in die Bereitschaftswarteschlange verantwortlich. Die Anzahl ist konfigurierbar, normalerweise reicht 2 aus, da die Nachrichten in der zset-Struktur nach Zeit geordnet sind

4.dq-consume-. N: Verantwortlich für das Lesen von Nachrichten aus der Bereitschaftswarteschlange und die Benachrichtigung der entsprechenden Fallback-Schnittstelle. Die Nummer ist konfigurierbar.

5.dq-redis-checker: Verantwortlich für die Überprüfung des Dienststatus von Redis ein Alarm. E-Mail

6.dq-http-server: Bietet Web-Hintergrundschnittstelle für die Registrierung des Themas

Bereitstellung

Umgebungsabhängigkeit: PHP 5.4 + Sockets, Redis, PCntl, pdo_mysql-Erweiterung installieren

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

Das obige ist der detaillierte Inhalt vonWie wird die PHP-Verzögerungswarteschlange implementiert?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!