PHP 지연 대기열은 어떻게 구현됩니까?

Delay 대기열은 이름에서 알 수 있듯이 지연 기능이 있는 메시지 대기열입니다. 그렇다면 어떤 상황에서 그러한 대기열이 필요합니까?

1. 배경

먼저 비즈니스 시나리오를 살펴보겠습니다.

1. 멤버십 만료 전 리콜 공지

2. 주문 결제가 성공한 후 5분 후에 다운스트림 링크가 정상인지 확인합니다. 예를 들어 사용자가 멤버십을 구매한 후 다양한 멤버십 상태가 설정되어 있는지 확인합니다. 성공적으로

3 .환불 상태의 주문이 성공적으로 환불되었는지 정기적으로 확인하는 방법은 무엇입니까?

4. 알림 구현에 실패했습니다. 상대방이 응답할 때까지 1, 3, 5, 7분 후에 알림을 반복하시겠습니까?

일반적으로 위의 문제를 해결하는 가장 간단하고 직접적인 방법은 측정기를 정기적으로 스캔하는 것입니다.

테이블 스캐닝의 문제점은 다음과 같습니다.

1 테이블 스캐닝이 대량으로 데이터베이스에 연결되는 경우가 많습니다. 비정상적인 중단 및 더 많은 예외가 필요합니다. 처리에는 높은 프로그램 견고성이 필요합니다

2. 데이터 양이 많을 경우 지연이 많아 규정 내에서 처리를 완료할 수 없어 비즈니스에 영향을 미칩니다. 처리를 위해 여러 프로세스를 시작할 수 있지만 이로 인해 추가 유지 관리 비용이 발생하며 근본적으로 해결할 수 없습니다.

3. 각 기업은 자체 측정기 스캔 논리를 유지해야 합니다. 비즈니스가 증가하면 테이블 스캐닝 부분의 로직이 반복적으로 개발될 것으로 나타났으나 이는

지연 큐가 위의 요구 사항을 매우 잘 해결할 수 있음

#과 매우 유사합니다. 🎜🎜##🎜 🎜#two、Research

에서는 다음과 같이 시장에 있는 일부 오픈 소스 솔루션을 조사했습니다.

·#🎜🎜 #Youzan 기술: 유일한 원칙, 오픈 소스 코드가 없습니다

·

github 개인: https://github.com/ouqiang/delay-queue

1. Redis 구현을 기반으로 Redis를 하나만 구성할 수 있습니다. Redis가 중단되면 전체 서비스를 사용할 수 없으며 가용성이 떨어집니다.

2. 소비자 측에서 구현합니다. 풀 모델이며 액세스 비용이 높으며 각 프로젝트를 구현해야 합니다.

3. Star를 프로덕션 환경에 배치하는 것은 위험합니다. go 언어를 이해하지 못하면 문제가 발생하면 유지 관리가 어려울 것입니다

·

SchedulerX-Alibaba 오픈 소스: 기능은 매우 강력하지만 운영 및 유지 관리가 복잡하고 종속 구성 요소가 많으며 충분히 가볍지 않습니다

·

RabbitMQ- 지연된 작업: 자체적으로 지연 기능이 없습니다. 게다가 회사에서는 이 대기열을 배포하지 않았습니다. 지연 대기열을 만들기 위해 별도의 대기열을 배포하는 것은 약간 비용이 많이 들고, 이를 유지하려면 특별한 운영 및 유지 관리가 필요합니다. . 현재 팀은 지원하지 않습니다

기본적으로는 위의 이유로 직접 작성하려고 합니다. 저는 주로 프로젝트의 기본 redis zset 구조를 저장소로 사용합니다. PHP 언어로 구현되었습니다. 구현 원리는 Youzan 팀을 참조하세요: https://tech .youzan.com/queuing_delay/

관련 권장 사항: "

php tutorial#🎜🎜 #"

3. Goal# 🎜🎜#

·Lightweight: 더 적은 PHP 확장으로 직접 실행할 수 있으며 스울, 워크맨 등 네트워크 프레임워크를 도입할 필요가 없습니다 # 🎜🎜#

·

안정성: 마스터워크 아키텍처를 사용하여 마스터가 업무 처리를 하지 않지만 하위 프로세스 관리만 담당합니다. 하위 프로세스가 비정상적으로 종료되면 자동으로 시작됩니다. # 🎜🎜#·

가용성:

1. 지원 다중 인스턴스 배포, 각 인스턴스는 상태 비저장이며 한 인스턴스의 실패는 서비스에 영향을 미치지 않습니다

2. 구성 지원 다중 Redis, 하나의 Redis가 실패하면 일부 메시지만 영향을 받습니다 #🎜🎜 #

3. 비즈니스 측에서는 쉽게 접근할 수 있습니다. 백그라운드에서 관련 메시지 유형과 반환 인터페이스만 입력하면 됩니다. #🎜 🎜#·

확장성: 소비 과정에서 병목 현상이 발생하면 소비 과정 수를 늘리도록 구성할 수 있어 쓰기에 병목 현상이 있을 때 인스턴스 수를 늘려 쓰기 성능을 선형적으로 향상시킬 수 있습니다

#🎜 🎜#·

실시간: 특정 시간 오류가 허용됩니다.

·

메시지 삭제 지원: 비즈니스 사용자는 언제든지 지정된 메시지를 삭제할 수 있습니다.

·

메시지 전송 신뢰성: 메시지가 지연 대기열에 들어간 후 적어도 한 번은 소비되는 것이 보장됩니다.

·

쓰기 성능: qps>1000+

4. 아키텍처 설계 및 설명

#🎜🎜 # 전체 아키텍처

여기에 그림 설명 삽입

은 주로 6개의 모듈을 포함하는 마스터 작업 아키텍처 모드를 채택합니다.

1 .dq- mster: 하위 프로세스의 생성, 파괴, 재활용 및 신호 알림을 관리하는 주요 프로세스

2.dq-server: 메시지 쓰기, 읽기, 삭제 기능 및 Redis 연결 풀 유지 관리를 담당 # 🎜🎜#3.dq-timer-N: redis의 zset 구조에서 만료된 메시지를 검색하고 준비 대기열에 쓰는 작업을 담당합니다. 메시지가 zset에 있으므로 일반적으로 2이면 충분합니다. 구조는 시간순으로 정렬됩니다.

4.dq-consume-N: 준비 대기열에서 메시지를 읽고 해당 반환 인터페이스에 알리는 역할을 담당하며 숫자는 구성 가능합니다

#🎜 🎜#5. dq-redis-checker: redis의 서비스 상태를 확인하는 역할을 담당합니다. redis가 다운되면 알람 이메일을 보냅니다

6.dq-http-server: 주제 등록을 위한 웹 백그라운드 인터페이스를 제공합니다 #🎜 🎜#

5. 배포

환경 종속성: PHP 5.4+ 소켓, redis, pcntl, 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

위 내용은 PHP 지연 대기열은 어떻게 구현됩니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!