FastCacheQueue

EasySwoole FastCache組件在>= 1.2.1的時候新增類似· beanstalkd消息隊列 ·特性。

• 可以創(chuàng)建多個queue
• 支持延遲投遞
• 任務超時恢復執(zhí)行
• 任務重發(fā)執(zhí)行
• 任務最大重發(fā)次數(shù)
• 支持putJob、delayJob、releaseJob、reserveJob、buryJob、kickJob等命令

基本使用

服務注冊

更新后，EasySwoole\FastCache\CacheProcessConfig類多出以下方法

/** 設置進程最大內(nèi)存 默認512M */
public function setMaxMem(string $maxMem): void
/** 設置消息隊列保留時間 默認60s （取出任務后沒有及時確認會重新放回隊列） */
public function setQueueReserveTime(int $queueReserveTime): void
/** 設置消息隊列最大重發(fā)次數(shù) 默認10 達到次數(shù)后重發(fā)將會被丟棄 */
public function setQueueMaxReleaseTimes(int $queueMaxReleaseTimes): void

開始使用

下文示例代碼的Job和Cache都使用以下命名空間

use EasySwoole\FastCache\Cache;
use EasySwoole\FastCache\Job;

投遞任務

投遞成功之后 將會返回該任務的jobId。

沒有失敗情況，除非fastCache注冊注冊失敗。

$job = new Job();
$job->setData("siam"); // 任意類型數(shù)據(jù)
$job->setQueue("siam_queue");
$jobId = Cache::getInstance()->putJob($job);
var_dump($jobId);

取出任務

可以開啟自定義進程當消費者，循環(huán)監(jiān)聽隊列，執(zhí)行任務處理。

$job = Cache::getInstance()->getJob('siam_queue');// Job對象或者null
if ($job === null){
    echo "沒有任務\n";
}else{
    // 執(zhí)行業(yè)務邏輯
    var_dump($job);
    // 執(zhí)行完了要刪除或者重發(fā)，否則超時會自動重發(fā)
    Cache::getInstance()->deleteJob($job);
}

清空ready任務隊列

 var_dump(Cache::getInstance()->flushReadyJobQueue('siam_queue'));

 var_dump(Cache::getInstance()->jobQueueSize('siam_queue'));

延遲執(zhí)行任務

$job = new Job();
$job->setData("siam");
$job->setQueue("siam_queue_delay");
$job->setDelay(5);// 延時5s
$jobId = Cache::getInstance()->putJob($job);
var_dump($jobId);
// 馬上取會失敗 隔5s取才成功
$job = Cache::getInstance()->getJob('siam_queue_delay');
var_dump($job);

刪除任務

可以是由getJob取出的對象，也可以自己聲明Job對象，傳入JobId來刪除。

$job = new Job();
$job->setJobId(1);
$job->setQueue('siam_queue_delay');
Cache::getInstance()->deleteJob($job);

任務重發(fā)

任務執(zhí)行失敗，或者某些場景需要重新執(zhí)行，則可以重發(fā)。

重發(fā)時，可以指定是否延遲執(zhí)行。

// get出來的任務執(zhí)行失敗可以重發(fā)
$job = new Job();
$job->setData("siam");
$job->setQueue("siam_queue");
$jobId = Cache::getInstance()->putJob($job);

$job = Cache::getInstance()->getJob('siam_queue');

if ($job === null){
    echo "沒有任務\n";
}else{
    // 執(zhí)行業(yè)務邏輯
    $doRes = false;
    if (!$doRes){
        // 業(yè)務邏輯失敗,需要重發(fā)  
        // 如果延遲隊列需要馬上重發(fā),在這里需要清空delay屬性
        // $job->setDelay(0);
        // 如果普通隊列需要延遲重發(fā),則設置delay屬性
        // $job->setDelay(5);
        $res = Cache::getInstance()->releaseJob($job);
        var_dump($res);
    }else{
        // 執(zhí)行完了要刪除或者重發(fā)，否則超時會自動重發(fā)
        Cache::getInstance()->deleteJob($job);
    }
}

返回現(xiàn)在有什么隊列

$queues = Cache::getInstance()->jobQueues();
var_dump($queues);

返回某個隊列的長度

$queueSize = Cache::getInstance()->jobQueueSize("siam_queue");
$queueSize2 = Cache::getInstance()->jobQueueSize("siam_queue_delay");
var_dump($queueSize);
var_dump($queueSize2);

清空隊列 可指定名稱

// 清空全部
$res = Cache::getInstance()->flushJobQueue();
var_dump($res);

// 清空siam_queue隊列
$res = Cache::getInstance()->flushJobQueue('siam_queue');
var_dump($res);

將任務改為延遲狀態(tài)

//添加任務
$job = new Job();
$job->setData("LuffyQAQ");
$job->setQueue("LuffyQAQ_queue_delay");
$jobId = Cache::getInstance()->putJob($job);

//方法一 直接傳入jobId
$job->setJobId($jobId);
$job->setDelay(30);
var_dump(Cache::getInstance()->delayJob($job));

//方法二 取出任務
$job = Cache::getInstance()->getJob('LuffyQAQ_queue_delay');
$job->setDelay(30);
var_dump(Cache::getInstance()->delayJob($job));

//使用jobQueueSize查看隊列長度
$queueSize = Cache::getInstance()->jobQueueSize("LuffyQAQ_queue_delay");
var_dump($queueSize);

從延遲執(zhí)行隊列中拿取

//傳入隊列名
var_dump(Cache::getInstance()->getDelayJob('LuffyQAQ_queue_delay'));

清空delay任務隊列

 var_dump(Cache::getInstance()->flushDelayJobQueue('LuffyQAQ_queue_delay'));

 var_dump(Cache::getInstance()->jobQueueSize('LuffyQAQ_queue_delay'));

將任務改為保留狀態(tài)

//添加任務
$job = new Job();
$job->setData("LuffyQAQ");
$job->setQueue("LuffyQAQ_queue_reserve");
$jobId = Cache::getInstance()->putJob($job);

//方法一 直接傳入jobId
$job->setJobId($jobId);
var_dump(Cache::getInstance()->reserveJob($job));

//方法二 取出任務
$job = Cache::getInstance()->getJob('LuffyQAQ_queue_reserve');
var_dump(Cache::getInstance()->reserveJob($job));

//使用jobQueueSize查看隊列長度
$queueSize = Cache::getInstance()->jobQueueSize("LuffyQAQ_queue_reserve");
var_dump($queueSize);

從保留隊列中拿取

//傳入隊列名
var_dump(Cache::getInstance()->getReserveJob('LuffyQAQ_queue_reserve'));

清空reserve任務隊列

 var_dump(Cache::getInstance()->flushReserveJobQueue('LuffyQAQ_queue_reserve'));

 var_dump(Cache::getInstance()->jobQueueSize('LuffyQAQ_queue_reserve'));

將任務改為埋藏狀態(tài)

$job = new Job();
$job->setQueue('LuffyQAQ_queue_bury');
$job->setData('LuffyQAQ');
$jobId = Cache::getInstance()->putJob($job);
$job->setJobId($jobId);

var_dump(Cache::getInstance()->buryJob($job));

//使用jobQueueSize查看隊列長度
$queueSize = Cache::getInstance()->jobQueueSize("LuffyQAQ_queue_bury");
var_dump($queueSize);

從埋藏隊列中拿取

//傳入隊列名
var_dump(Cache::getInstance()->getBuryJob('LuffyQAQ_queue_bury'));

將埋藏隊列任務恢復到ready中

var_dump(Cache::getInstance()->kickJob($job));

清空bury任務隊列

 var_dump(Cache::getInstance()->flushBuryJobQueue('LuffyQAQ_queue_bury'));

 var_dump(Cache::getInstance()->jobQueueSize('LuffyQAQ_queue_bury'));