EasySwoole 基于Redis組件實(shí)現(xiàn)延遲隊(duì)列

介紹

在用戶要支付訂單的時(shí)候，如果超過(guò)30分鐘未支付，會(huì)把訂單關(guān)掉。當(dāng)然我們可以做一個(gè)定時(shí)任務(wù)，每個(gè)一段時(shí)間來(lái)掃描未支付的訂單，如果該訂單超過(guò)支付時(shí)間就關(guān)閉，但是在數(shù)據(jù)量小的時(shí)候并沒(méi)有什么大的問(wèn)題，但是數(shù)據(jù)量一大輪訓(xùn)數(shù)據(jù)庫(kù)的方式就會(huì)變得特別耗資源。當(dāng)面對(duì)千萬(wàn)級(jí)、上億級(jí)數(shù)據(jù)量時(shí)，本身寫入的IO就比較高，導(dǎo)致長(zhǎng)時(shí)間查詢或者根本就查不出來(lái)，更別說(shuō)分庫(kù)分表以后了。

使用延遲隊(duì)列解決的痛點(diǎn)無(wú)非是

1. 實(shí)現(xiàn)了數(shù)據(jù)延遲
2. 數(shù)據(jù)攤開(kāi)(仔細(xì)去理解)

知識(shí)點(diǎn)

1. redis有序集合
2. EasySwoole Redis協(xié)程客戶端

案例

生成訂單id ---> 扔到延遲隊(duì)列 ---> 延遲隊(duì)列消費(fèi)進(jìn)程不停獲取30分鐘前的訂單滿足條件的訂單 ---> 處理訂單

直接上代碼

EasySwooleEvent.php 注冊(cè)redis連接池、注冊(cè)延遲隊(duì)列消費(fèi)進(jìn)程

<?php
namespace EasySwoole\EasySwoole;

use App\Process\Consumer;
use EasySwoole\EasySwoole\Swoole\EventRegister;
use EasySwoole\EasySwoole\AbstractInterface\Event;
use EasySwoole\Http\Request;
use EasySwoole\Http\Response;
use EasySwoole\Pool\Manager;
use EasySwoole\Redis\Config\RedisConfig;
use App\RedisPool\RedisPool;
use EasySwoole\Pool\Config;
class EasySwooleEvent implements Event
{

    public static function initialize()
    {
        // TODO: Implement initialize() method.
        date_default_timezone_set('Asia/Shanghai');
    }

    public static function mainServerCreate(EventRegister $register)
    {

        //TODO:: 注冊(cè)redis連接池
        $config = new Config();
        $redisConfig1 = new RedisConfig([
            'host'      => '127.0.0.1',
            'port'      => '6379'
        ]);

        // 這里的redis連接池看文檔配吧
        Manager::getInstance()->register(new RedisPool($config,$redisConfig1),'redis');

        //TODO:: 延遲隊(duì)列消費(fèi)進(jìn)程
        $processConfig= new \EasySwoole\Component\Process\Config();
        $processConfig->setProcessName('testProcess');

        \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new Consumer($processConfig));
    }

    public static function onRequest(Request $request, Response $response): bool
    {
        // TODO: Implement onRequest() method.

        return true;
    }

    public static function afterRequest(Request $request, Response $response): void
    {
        // TODO: Implement afterAction() method.
    }
}

扔到延遲隊(duì)列

<?php
namespace App\HttpController;

use EasySwoole\Http\AbstractInterface\Controller;
use EasySwoole\Pool\Manager;

class Index extends Controller
{

    function index()
    {
        /** @var $redis \EasySwoole\Redis\Redis*/
        $orderId = date('YmdHis', time());
        $redis = Manager::getInstance()->get('redis')->getObj();
        $res = $redis->zAdd('delay_queue_test1', time(), $orderId);
        if ($res) {
            $this->writeJson(200, '訂單添加成功:'.$orderId);
        }
    }

}

延遲隊(duì)列消費(fèi)進(jìn)程

<?php
namespace App\Process;

use EasySwoole\Component\Process\AbstractProcess;
use EasySwoole\Pool\Manager;
use Swoole\Coroutine;

class Consumer extends AbstractProcess {
    protected function run($arg)
    {
        go(function (){
            while (true) {

                //TODO:: 拿到redis
                /** @var $redis \EasySwoole\Redis\Redis*/
                $redis = Manager::getInstance()->get('redis')->defer();

                //TODO:: 從有序集合中拿到三秒(模擬30分鐘)以前的訂單
                $orderIds = $redis->zRangeByScore('delay_queue_test1', 0, time()-3, ['withscores' => TRUE]);

                if (empty($orderIds)) {
                    Coroutine::sleep(1);
                    continue;
                }

                //TODO::拿出后立馬刪除
                $redis->zRem('delay_queue_test1', ...$orderIds);

                foreach ($orderIds as $orderId)
                {
                    var_dump($orderId);

                    //TODO::判斷此訂單30分鐘后，是否仍未完成，做相應(yīng)處理
                }
            }
        });
    }

}

測(cè)試

請(qǐng)求index/index 投遞訂單到延遲隊(duì)列

?  ~ curl 127.0.0.1:9501/index/index
{"code":200,"result":"訂單添加成功:20200422004046","msg":null}%

等3s看終端是否輸出

?  php easyswoole.php start
  ______                          _____                              _
 |  ____|                        / ____|                            | |
 | |__      __ _   ___   _   _  | (___   __      __   ___     ___   | |   ___
 |  __|    / _` | / __| | | | |  \___ \  \ \ /\ / /  / _ \   / _ \  | |  / _ \
 | |____  | (_| | \__ \ | |_| |  ____) |  \ V  V /  | (_) | | (_) | | | |  __/
 |______|  \__,_| |___/  \__, | |_____/    \_/\_/    \___/   \___/  |_|  \___|
                          __/ |
                         |___/
main server                   SWOOLE_WEB
listen address                0.0.0.0
listen port                   9501
ip@en0                        192.168.43.57
worker_num                    8
reload_async                  true
max_wait_time                 3
pid_file                      /Users/xx/sites/easyswoole/Temp/pid.pid
log_file                      /Users/xx/sites/easyswoole/Log/swoole.log
user                          xx
daemonize                     false
swoole version                4.4.15
php version                   7.2.18
easy swoole                   3.3.7
develop/produce               develop
temp dir                      /Users/xx/sites/easyswoole/Temp
log dir                       /Users/xx/sites/easyswoole/Log

string(14) "20200422004046"

總結(jié)

這只是一個(gè)思路，大家可以根據(jù)實(shí)際業(yè)務(wù)做不同調(diào)整