PHP implementiert eine leichte Verzögerungswarteschlange (Multithreading) basierend auf Redis

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 Tabellenscannen sind:

1. Der Tabellenscan ist über einen längeren Zeitraum 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, ist die Verzögerung hoch und die Verarbeitung kann nicht innerhalb der Vorschriften abgeschlossen werden, was sich 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:

1 Youzan-Technologie: nur Prinzipien, kein Open-Source-Code

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

(1) Basierend auf der Redis-Implementierung kann nur ein Redis konfiguriert werden. Wenn Redis hängt, ist der gesamte Dienst nicht verfügbar Die Verfügbarkeit wird fast sein

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

(3) Es gibt nicht viele Leute Wenn es in Star verwendet wird, ist es riskant, es in die Produktionsumgebung zu integrieren. In Verbindung mit dem mangelnden Verständnis der Go-Sprache ist es schwierig, es aufrechtzuerhalten, wenn ein Problem auftritt

3 Open Source: sehr leistungsstark, aber komplex in Betrieb und Wartung, mit vielen abhängigen Komponenten und nicht leicht genug

4 .RabbitMQ-verzögerte Aufgabe: Sie hat selbst keine Verzögerungsfunktion. Sie muss von implementiert werden Darüber hinaus hat das Unternehmen diese Warteschlange nicht separat bereitgestellt, um eine Verzögerungswarteschlange zu erstellen, und es erfordert auch eine spezielle Bedienung und Wartung, die das Team derzeit durchführt nicht unterstützt

Grundsätzlich habe ich vor, selbst eines zu schreiben. Als Speicher verwende ich grundsätzlich die Zset-Struktur von Redis , wenden Sie sich bitte an das Youzan-Team: https://tech.youzan.com/queuing_delay/

Die gesamte Verzögerungswarteschlange besteht hauptsächlich aus 4 Teilen

JobPool wird zum Speichern der Metainformationen verwendet aller Berufe.

DelayBucket ist eine Reihe geordneter Warteschlangen mit der Dimension Zeit, die zum Speichern aller Jobs verwendet werden, die verzögert werden müssen (hier werden nur Job-IDs gespeichert).

Der Timer ist dafür verantwortlich, jeden Bucket in Echtzeit zu scannen und Jobs, deren Verzögerungszeit größer oder gleich der aktuellen Zeit ist, in die entsprechende Bereitschaftswarteschlange zu platzieren.

ReadyQueue speichert Jobs im Status „Bereit“ (hier werden nur JobIds gespeichert) zur Nutzung durch Verbraucherprogramme.

Nachrichtenstruktur

Jeder Job muss die folgenden Attribute enthalten:

Thema: Jobtyp. Es kann als spezifischer Firmenname verstanden werden.

id: Die eindeutige Kennung des Jobs. Wird zum Abrufen und Löschen angegebener Jobinformationen verwendet.

delayTime: jod verzögerte Ausführungszeit, 13-stelliger Zeitstempel

ttr (time-to-run): Zeitüberschreitung bei der Auftragsausführung.

Körper: Der Inhalt des Jobs, damit Verbraucher bestimmte Geschäftsabwicklungen durchführen können, gespeichert im JSON-Format.

Für die gleiche Art von Themenverzögerungszeit ist ttr im Allgemeinen festgelegt und die Jobattribute können vereinfacht werden

1.Thema: Jobtyp. Es kann als spezifischer Firmenname verstanden werden

2.id: die eindeutige Kennung des Jobs. Wird zum Abrufen und Löschen angegebener Jobinformationen verwendet.

3.body: Der Inhalt des Jobs, damit Verbraucher bestimmte Geschäftsabwicklungen durchführen können, gespeichert im JSON-Format.

Verzögerungszeit, TTR werden im Topicadmin-Hintergrund konfiguriert

3. Ziel

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

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, und das wird auch der Fall sein wird automatisch aufgerufen, wenn der untergeordnete Prozess abnormal beendet wird

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, wenn ein Redis ausfällt. Betrifft nur einige Nachrichten

3. Die Geschäftsseite hat einfachen Zugriff und muss nur die relevanten Nachrichtentypen und Rückrufschnittstellen ausfüllen der Hintergrund

Skalierbarkeit: Wenn es einen Engpass im Verbrauchsprozess gibt, können Sie ihn so konfigurieren, dass der Verbrauch erhöht wird. Die Anzahl der Prozesse kann beim Schreiben erhöht werden Die Leistung kann linear verbessert werden.

Echtzeit: Ein bestimmter Zeitfehler ist zulässig.

Löschung von Nachrichten unterstützen: 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

采用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. Wenn die Benachrichtigungsschnittstelle innerhalb des Timeout-Zeitraums aufgerufen wird und keine Antwort eingeht, wird die Benachrichtigung als fehlgeschlagen betrachtet. Das System stellt die Daten erneut in die Warteschlange und benachrichtigt erneut auf eine maximale Benachrichtigung von 10 Mal (kann in der Dqconf.php-Datei $ notify_exp_nums geändert werden) Das Benachrichtigungsintervall beträgt 2n+1. Wenn die Benachrichtigung beispielsweise zum ersten Mal 1 Minute lang fehlschlägt, bis nach 3 Minuten zum zweiten Mal eine Antwort eingeht, Das System verwirft es automatisch, wenn die maximale Anzahl an Benachrichtigungen überschritten wird, und sendet gleichzeitig eine E-Mail-Benachrichtigung

2. Online-Redis wird alle 1 Sekunde gespeichert, und es kann Fälle geben, in denen Daten für 1 verloren gehen Zweitens. In diesem Fall können Sie es manuell wiederherstellen, indem Sie die Protokolle request_ymd.txt und notify_ymd.txt vergleichen

3.

ps : Netzwerkjitter ist unvermeidlich. Wenn die Benachrichtigungsschnittstelle Kerndienste umfasst, muss sie idempotent sein! !

10. Online-Situation

Zwei Instanzen wurden online bereitgestellt, eine in jedem Computerraum, 4 Redis mit insgesamt 16G Speicher für die Speicherung, und der Dienst wurde bereitgestellt Die Entwicklung läuft seit mehreren Monaten stabil, alle Indikatoren entsprechen den Erwartungen.

Hauptzugangsgeschäft:

·10-minütige Rückrufbenachrichtigung

·Entschädigung, wenn der Aufruf der Schnittstelle abläuft oder fehlschlägt

·Rückrufbenachrichtigung 3 Tage vor Ablauf der Mitgliedschaft

11. Mängel und Aussichten

1. Da dem vom Team verwendeten Image die Libevent-Erweiterung fehlt, basiert der DQ-Server auf dem Select-Modell und es kann zu Leistungsengpässen kommen geändert, um in Zukunft auf dem libevent-Ereignismodell zu basieren, um die Parallelitätsleistung zu verbessern.

2. Timer und Verbraucher werden derzeit mithilfe mehrerer Prozesse implementiert. Diese Granularität fühlt sich etwas grob an. Sie können die dynamische Erstellung der Anzahl von Threads in Betracht ziehen, um die Verbraucherleistung zu verbessern und sicherzustellen das größte Ausmaß.

3.dq-server und redis werden synchron aufgerufen, was ebenfalls einen Leistungsengpass darstellt. Wir planen, es asynchron basierend auf swoole_redis zu verarbeiten.

PHP-Website für Chinesisch, es gibt viele kostenlose PHP-Video-Tutorials, jeder ist herzlich willkommen, es zu lernen!

Dieser Artikel wurde reproduziert von: https://www.jianshu.com/p/58f10ac42162