摘要：如果任務(wù)沒(méi)有在規(guī)定時(shí)間內(nèi)完成，那么該有序集合的任務(wù)將會(huì)被重新放入隊(duì)列中。這兩個(gè)進(jìn)程操縱了三個(gè)隊(duì)列，其中一個(gè)，負(fù)責(zé)即時(shí)任務(wù)，兩個(gè)，負(fù)責(zé)延時(shí)任務(wù)與待處理任務(wù)。如果任務(wù)執(zhí)行成功，就會(huì)刪除中的任務(wù)，否則會(huì)被重新放入隊(duì)列中。

在實(shí)際的項(xiàng)目開(kāi)發(fā)中，我們經(jīng)常會(huì)遇到需要輕量級(jí)隊(duì)列的情形，例如發(fā)短信、發(fā)郵件等，這些任務(wù)不足以使用 kafka、RabbitMQ 等重量級(jí)的消息隊(duì)列，但是又的確需要異步、重試、并發(fā)控制等功能。通常來(lái)說(shuō)，我們經(jīng)常會(huì)使用 Redis、Beanstalk、Amazon SQS 來(lái)實(shí)現(xiàn)相關(guān)功能，laravel 為此對(duì)不同的后臺(tái)隊(duì)列服務(wù)提供統(tǒng)一的 API，本文將會(huì)介紹應(yīng)用最為廣泛的 redis 隊(duì)列。

在講解 laravel 的隊(duì)列服務(wù)之前，我們要先說(shuō)說(shuō)基于 redis 的隊(duì)列服務(wù)。首先，redis設(shè)計(jì)用來(lái)做緩存的，但是由于它自身的某種特性使得它可以用來(lái)做消息隊(duì)列

redis 做消息隊(duì)列的特性例如FIFO（先入先出）很容易實(shí)現(xiàn)，只需要一個(gè) list 對(duì)象從頭取數(shù)據(jù)，從尾部塞數(shù)據(jù)即可。

相關(guān)的命令：（1）左側(cè)入右側(cè)出：lpush/rpop；（2）右側(cè)入左側(cè)出：rpush/lpop。

這個(gè)簡(jiǎn)單的消息隊(duì)列很容易實(shí)現(xiàn)。

有些任務(wù)場(chǎng)景，并不需要任務(wù)立刻執(zhí)行，而是需要延遲執(zhí)行；有些任務(wù)很重要，需要在任務(wù)失敗的時(shí)候重新嘗試。這些功能僅僅依靠 list 是無(wú)法完成的。這個(gè)時(shí)候，就需要 redis 的有序集合。

Redis 有序集合和 Redis 集合類(lèi)似，是不包含相同字符串的合集。它們的差別是，每個(gè)有序集合的成員都關(guān)聯(lián)著一個(gè)評(píng)分 score，這個(gè)評(píng)分用于把有序集合中的成員按最低分到最高分排列。

單看有序集合和延遲任務(wù)并無(wú)關(guān)系，但是可以將有序集合的評(píng)分 score 設(shè)置為延時(shí)任務(wù)開(kāi)啟的時(shí)間，之后輪詢(xún)這個(gè)有序集合，將到期的任務(wù)拿出來(lái)進(jìn)行處理，這樣就實(shí)現(xiàn)了延遲任務(wù)的功能。

對(duì)于重要的需要重試的任務(wù)，在任務(wù)執(zhí)行之前，會(huì)將該任務(wù)放入有序集合中，設(shè)置任務(wù)最長(zhǎng)的執(zhí)行時(shí)間。若任務(wù)順利執(zhí)行完畢，該任務(wù)會(huì)在有序集合中刪除。如果任務(wù)沒(méi)有在規(guī)定時(shí)間內(nèi)完成，那么該有序集合的任務(wù)將會(huì)被重新放入隊(duì)列中。

相關(guān)命令：

(1) ZADD 添加一個(gè)或多個(gè)成員到有序集合，或者如果它已經(jīng)存在更新其分?jǐn)?shù)。

(2) ZRANGEBYSCORE 按分?jǐn)?shù)返回一個(gè)成員范圍的有序集合。

(3) ZREMRANGEBYRANK 在給定的索引之內(nèi)刪除所有成員的有序集合。

隊(duì)列服務(wù)的任務(wù)調(diào)度過(guò)程如下：

laravel 的隊(duì)列服務(wù)由兩個(gè)進(jìn)程控制，一個(gè)是生產(chǎn)者，一個(gè)是消費(fèi)者。這兩個(gè)進(jìn)程操縱了 redis 三個(gè)隊(duì)列，其中一個(gè) List，負(fù)責(zé)即時(shí)任務(wù)，兩個(gè) Zset，負(fù)責(zé)延時(shí)任務(wù)與待處理任務(wù)。

生產(chǎn)者負(fù)責(zé)向 redis 推送任務(wù)，如果是即時(shí)任務(wù)，默認(rèn)就會(huì)向 queue:default 推送；如果是延時(shí)任務(wù)，就會(huì)向 queue:default:delayed 推送。

消費(fèi)者輪詢(xún)兩個(gè)隊(duì)列，不斷的從隊(duì)列中取出任務(wù)，先把任務(wù)放入 queue:default:reserved 中，再執(zhí)行相關(guān)任務(wù)。如果任務(wù)執(zhí)行成功，就會(huì)刪除 queue:default:reserved 中的任務(wù)，否則會(huì)被重新放入 queue:default:delayed 隊(duì)列中。

任務(wù)分發(fā)流程：

任務(wù)處理器運(yùn)作：

"redis" => [
        "driver" => "redis",
        "connection" => "default",
        "queue" => "default",
        "retry_after" => 90,
    ],

在config/queue.php中進(jìn)行配置
一般來(lái)說(shuō)，默認(rèn)的 redis 配置如上，connection 是 database 中 redis 的連接名稱(chēng)；queue 是 redis 中的隊(duì)列名稱(chēng)，值得注意的是，如果使用的是 redis 集群的話(huà)，這個(gè)需要使用 key hash tag，也就是 {default};當(dāng)任務(wù)運(yùn)行超過(guò) retry_after 這個(gè)時(shí)間后，該任務(wù)會(huì)被重新放入隊(duì)列當(dāng)中。

任務(wù)類(lèi)的結(jié)構(gòu)很簡(jiǎn)單，一般來(lái)說(shuō)只會(huì)包含一個(gè)讓隊(duì)列用來(lái)調(diào)用此任務(wù)的 handle 方法。

如果想要使得任務(wù)被推送到隊(duì)列中，而不是同步執(zhí)行，那么需要實(shí)現(xiàn) IlluminateContractsQueueShouldQueue 接口。

如果想要讓任務(wù)推送到特定的連接中，例如 redis 或者 sqs，那么需要設(shè)置 conneciton 變量。

如果想要讓任務(wù)推送到特定的隊(duì)列中去，可以設(shè)置 queue 變量。

如果想要讓任務(wù)延遲推送，那么需要設(shè)置 delay 變量。

如果想要設(shè)置任務(wù)至多重試的次數(shù)，可以使用 tries 變量；

如果想要設(shè)置任務(wù)可以運(yùn)行的最大秒數(shù)，那么可以使用 timeout 參數(shù)。

如果想要手動(dòng)訪(fǎng)問(wèn)隊(duì)列，可以使用 trait : IlluminateQueueInteractsWithQueue。

任務(wù)的分發(fā)
分發(fā)服務(wù)
寫(xiě)好任務(wù)類(lèi)后，就能通過(guò) dispatch 輔助函數(shù)來(lái)分發(fā)它了。唯一需要傳遞給 dispatch 的參數(shù)是這個(gè)任務(wù)類(lèi)的實(shí)例：

class PodcastController extends Controller
{
    public function store(Request $request)
    {
        // 創(chuàng)建播客...

        ProcessPodcast::dispatch($podcast);
    }
}

如果想延遲執(zhí)行一個(gè)隊(duì)列中的任務(wù)，可以用任務(wù)實(shí)例的 delay 方法。

 ProcessPodcast::dispatch($podcast)
                ->delay(Carbon::now()->addMinutes(10));

通過(guò)推送任務(wù)到不同的隊(duì)列，可以給隊(duì)列任務(wù)分類(lèi)，甚至可以控制給不同的隊(duì)列分配多少任務(wù)。要指定隊(duì)列的話(huà)，就調(diào)用任務(wù)實(shí)例的 onQueue 方法：

ProcessPodcast::dispatch($podcast)->onQueue("processing");

如果使用了多個(gè)隊(duì)列連接，可以將任務(wù)推到指定連接。要指定連接的話(huà)，可以在分發(fā)任務(wù)的時(shí)候使用 onConnection 方法：

ProcessPodcast::dispatch($podcast)->onConnection("redis
");

參考資源

KingFer