純php+socket實現延遲隊列

時間 2019-11-13

標籤 php+socket php socket 實現延遲隊列欄目 PHP 简体版

原文原文鏈接

延遲隊列，顧名思義它是一種帶有延遲功能的消息隊列。那麼，是在什麼場景下我才須要這樣的隊列呢？php

1、背景

先看看一下業務場景：mysql

1.會員過時前3天發送召回通知
2.訂單支付成功後，5分鐘後檢測下游環節是否都正常，好比用戶購買會員後，各類會員狀態是否都設置成功
3.如何按期檢查處於退款狀態的訂單是否已經退款成功？
4.實現通知失敗，1，3，5，7分鐘重複通知，直到對方回覆？

一般解決以上問題，最簡單直接的辦法就是定時去掃表。git

掃表存在的問題是：github

1.掃表與數據庫長時間鏈接，在數量量大的狀況容易出現鏈接異常中斷，須要更多的異常處理，對程序健壯性要求高
2.在數據量大的狀況下延時較高，規定內處理不完，影響業務，雖然能夠啓動多個進程來處理，這樣會帶來額外的維護成本，不能從根本上解決。
3.每一個業務都要維護一個本身的掃表邏輯。當業務愈來愈多時，發現掃表部分的邏輯會重複開發，可是很是相似

延時隊列能對於上述需求能很好的解決web

2、調研

調研了市場上一些開源的方案，如下：redis

1.有贊科技：只有原理，沒有開源代碼sql

2.github我的的：https://github.com/ouqiang/delay-queue數據庫

1.基於redis實現，redis只能配置一個,若是redis掛了整個服務不可用，可用性差點
2.消費端實現的是拉模式，接入成本大，每一個項目都得去實現一遍接入代碼
3.在star使用的人數很少，放在生產環境，存在風險，加之對go語言不瞭解，出了問題難以維護

3.SchedulerX-阿里開源的：功能很強大，可是運維複雜，依賴組件多，不夠輕量性能優化
4.RabbitMQ-延時任務: 自己沒有延時功能，須要藉助一特性本身實現，並且公司沒有部署這個隊列，去單獨部署一個這個來作延時隊列成本有點高，並且還須要專門的運維來維護，目前團隊不支持swoole

基本以上緣由打算本身寫一個，日常使用php多，項目基本redis的zset結構做爲存儲，用php語言實現，實現原理參考了有贊團隊：https://tech.youzan.com/queuing_delay/

3、目標

輕量級：有較少的php的拓展就能直接運行，不須要引入網絡框架，好比swoole，workman之類的
穩定性：採用master-work架構，master不作業務處理，只負責管理子進程，子進程異常退出時自動拉起
可用性：
- 1.支持多實例部署，每一個實例無狀態，一個實例掛掉不影響服務
- 2.支持配置多個redis，一個redis掛了隻影響部分消息
- 3.業務方接入方便，在後臺只需填寫相關消息類型和回掉接口
拓展性: 當消費進程存在瓶頸時，能夠配置加大消費進程數，當寫入存在瓶頸時，可增長實例數寫入性能可線性提升
實時性：容許存在必定的時間偏差。
支持消息刪除：業務使用方，能夠隨時刪除指定消息。
消息傳輸可靠性：消息進入到延遲隊列後，保證至少被消費一次。
寫入性能：qps>1000+

4、架構設計與說明

整體架構

採用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

5、部署

環境依賴：PHP 5.4+ 安裝sockets，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 'DqLoader.php';
date_default_timezone_set("PRC");
//可配置多個
$server=array(
    '127.0.0.1:6789',
);
$dqClient = new DqClient();
$dqClient->addServer($server);

$topic ='order_openvip_checker'; //topic在後臺註冊
$id = uniqid();
$data=array(
    'id'=>$id,
    'body'=>array(
        'a'=>1,
        'b'=>2,
        'c'=>3,
        'ext'=>str_repeat('a',64),
    ),
    //可選，設置後以這個通知時間爲準，默認延時時間在註冊topic的時候指定
    'fix_time'=>date('Y-m-d 23:50:50'),
);

//添加
$boolRet = $dqClient->add($topic, $data);
echo 'add耗時:'.(msectime() - $time)."ms\n";
//查詢
$time = msectime();
$result = $dqClient->get($topic, $id);
echo 'get耗時:'.(msectime() - $time)."ms\n";

//刪除
$time = msectime();
$boolRet = $dqClient->del($topic,$id);
echo 'del耗時:'.(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

6、性能測試

須要安裝pthreads拓展：

測試原理：使用多線程模擬併發，在1s內能成功返回請求成功的個數

php DqBench  concurrency  requests
concurrency:併發數
requests： 每一個併發產生的請求數

測試環境：內存 8G ，8核cpu，2個redis和1個dq-server 部署在一個機器上，數據包64字節
qps：2400

7、值得一提的性能優化點：

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 ，發現某個系統函數調用是其餘的數倍，可能大機率程序存在問題

8、異常處理

若是調用通知接口在超時時間內，沒有收到回覆認爲通知失敗，系統會從新把數據放入隊列，從新通知，系統默認最大通知10次(能夠在Dqconf.php文件中修改$notify_exp_nums)通知間隔爲2n+1，好比第一次1分鐘，通知失敗，第二次3分鐘後，直到收到回覆，超出最大通知次數後系統自動丟棄，同時發郵件通知

ps:網絡抖動在所不免，通知接口若是涉及到核心的服務,必定要保證冪等！！