EasySwoole 基于Redis組件實現延遲隊列

介紹

在用戶要支付訂單的時候，如果超過30分鐘未支付，會把訂單關掉。當然我們可以做一個定時任務，每個一段時間來掃描未支付的訂單，如果該訂單超過支付時間就關閉，但是在數據量小的時候并沒有什么大的問題，但是數據量一大輪訓數據庫的方式就會變得特別耗資源。當面對千萬級、上億級數據量時，本身寫入的IO就比較高，導致長時間查詢或者根本就查不出來，更別說分庫分表以后了。

使用延遲隊列解決的痛點無非是

1. 實現了數據延遲
2. 數據攤開(仔細去理解)

知識點

1. redis有序集合
2. EasySwoole Redis協程客戶端

案例

生成訂單id ---> 扔到延遲隊列 ---> 延遲隊列消費進程不停獲取30分鐘前的訂單滿足條件的訂單 ---> 處理訂單

直接上代碼

EasySwooleEvent.php 注冊redis連接池、注冊延遲隊列消費進程

<?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:: 注冊redis連接池
        $config = new Config();
        $redisConfig1 = new RedisConfig([
            'host'      => '127.0.0.1',
            'port'      => '6379'
        ]);

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

        //TODO:: 延遲隊列消費進程
        $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.
    }
}

扔到延遲隊列

<?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);
        }
    }

}

延遲隊列消費進程

<?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分鐘后，是否仍未完成，做相應處理
                }
            }
        });
    }

}

測試

請求index/index 投遞訂單到延遲隊列

?  ~ 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"

總結

這只是一個思路，大家可以根據實際業務做不同調整