資訊專欄INFORMATION COLUMN
摘要：如果任務(wù)沒有在規(guī)定時間內(nèi)完成，那么該有序集合的任務(wù)將會被重新放入隊列中。這兩個進程操縱了三個隊列，其中一個，負責即時任務(wù)，兩個，負責延時任務(wù)與待處理任務(wù)。如果任務(wù)執(zhí)行成功，就會刪除中的任務(wù)，否則會被重新放入隊列中。
在實際的項目開發(fā)中，我們經(jīng)常會遇到需要輕量級隊列的情形，例如發(fā)短信、發(fā)郵件等，這些任務(wù)不足以使用 kafka、RabbitMQ 等重量級的消息隊列，但是又的確需要異步、重試、并發(fā)控制等功能。通常來說，我們經(jīng)常會使用 Redis、Beanstalk、Amazon SQS 來實現(xiàn)相關(guān)功能，laravel 為此對不同的后臺隊列服務(wù)提供統(tǒng)一的 API，本文將會介紹應(yīng)用最為廣泛的 redis 隊列。
在講解 laravel 的隊列服務(wù)之前，我們要先說說基于 redis 的隊列服務(wù)。首先，redis設(shè)計用來做緩存的，但是由于它自身的某種特性使得它可以用來做消息隊列
redis 隊列的數(shù)據(jù)結(jié)構(gòu) List 鏈表redis 做消息隊列的特性例如FIFO（先入先出）很容易實現(xiàn)，只需要一個 list 對象從頭取數(shù)據(jù)，從尾部塞數(shù)據(jù)即可。
相關(guān)的命令：（1）左側(cè)入右側(cè)出：lpush/rpop；（2）右側(cè)入左側(cè)出：rpush/lpop。
這個簡單的消息隊列很容易實現(xiàn)。
Zset 有序集合有些任務(wù)場景，并不需要任務(wù)立刻執(zhí)行，而是需要延遲執(zhí)行；有些任務(wù)很重要，需要在任務(wù)失敗的時候重新嘗試。這些功能僅僅依靠 list 是無法完成的。這個時候，就需要 redis 的有序集合。
Redis 有序集合和 Redis 集合類似，是不包含相同字符串的合集。它們的差別是，每個有序集合的成員都關(guān)聯(lián)著一個評分 score，這個評分用于把有序集合中的成員按最低分到最高分排列。
單看有序集合和延遲任務(wù)并無關(guān)系，但是可以將有序集合的評分 score 設(shè)置為延時任務(wù)開啟的時間，之后輪詢這個有序集合，將到期的任務(wù)拿出來進行處理，這樣就實現(xiàn)了延遲任務(wù)的功能。
對于重要的需要重試的任務(wù)，在任務(wù)執(zhí)行之前，會將該任務(wù)放入有序集合中，設(shè)置任務(wù)最長的執(zhí)行時間。若任務(wù)順利執(zhí)行完畢，該任務(wù)會在有序集合中刪除。如果任務(wù)沒有在規(guī)定時間內(nèi)完成，那么該有序集合的任務(wù)將會被重新放入隊列中。
相關(guān)命令：
(1) ZADD 添加一個或多個成員到有序集合，或者如果它已經(jīng)存在更新其分數(shù)。
(2) ZRANGEBYSCORE 按分數(shù)返回一個成員范圍的有序集合。
(3) ZREMRANGEBYRANK 在給定的索引之內(nèi)刪除所有成員的有序集合。
laravel 隊列服務(wù)的任務(wù)調(diào)度隊列服務(wù)的任務(wù)調(diào)度過程如下：
laravel 的隊列服務(wù)由兩個進程控制，一個是生產(chǎn)者，一個是消費者。這兩個進程操縱了 redis 三個隊列，其中一個 List，負責即時任務(wù)，兩個 Zset，負責延時任務(wù)與待處理任務(wù)。
生產(chǎn)者負責向 redis 推送任務(wù)，如果是即時任務(wù)，默認就會向 queue:default 推送；如果是延時任務(wù)，就會向 queue:default:delayed 推送。
消費者輪詢兩個隊列，不斷的從隊列中取出任務(wù)，先把任務(wù)放入 queue:default:reserved 中，再執(zhí)行相關(guān)任務(wù)。如果任務(wù)執(zhí)行成功，就會刪除 queue:default:reserved 中的任務(wù)，否則會被重新放入 queue:default:delayed 隊列中。
laravel 隊列服務(wù)的總體流程任務(wù)分發(fā)流程：
任務(wù)處理器運作：
"redis" => [
"driver" => "redis",
"connection" => "default",
"queue" => "default",
"retry_after" => 90,
],
在config/queue.php中進行配置
一般來說，默認的 redis 配置如上，connection 是 database 中 redis 的連接名稱；queue 是 redis 中的隊列名稱，值得注意的是，如果使用的是 redis 集群的話，這個需要使用 key hash tag，也就是 {default};當任務(wù)運行超過 retry_after 這個時間后，該任務(wù)會被重新放入隊列當中。
任務(wù)類的結(jié)構(gòu)很簡單，一般來說只會包含一個讓隊列用來調(diào)用此任務(wù)的 handle 方法。
如果想要使得任務(wù)被推送到隊列中，而不是同步執(zhí)行，那么需要實現(xiàn) IlluminateContractsQueueShouldQueue 接口。
如果想要讓任務(wù)推送到特定的連接中，例如 redis 或者 sqs，那么需要設(shè)置 conneciton 變量。
如果想要讓任務(wù)推送到特定的隊列中去，可以設(shè)置 queue 變量。
如果想要讓任務(wù)延遲推送，那么需要設(shè)置 delay 變量。
如果想要設(shè)置任務(wù)至多重試的次數(shù)，可以使用 tries 變量；
如果想要設(shè)置任務(wù)可以運行的最大秒數(shù)，那么可以使用 timeout 參數(shù)。
如果想要手動訪問隊列，可以使用 trait : IlluminateQueueInteractsWithQueue。
任務(wù)的分發(fā)
分發(fā)服務(wù)
寫好任務(wù)類后，就能通過 dispatch 輔助函數(shù)來分發(fā)它了。唯一需要傳遞給 dispatch 的參數(shù)是這個任務(wù)類的實例：
class PodcastController extends Controller
{
public function store(Request $request)
{
// 創(chuàng)建播客...
ProcessPodcast::dispatch($podcast);
}
}
如果想延遲執(zhí)行一個隊列中的任務(wù)，可以用任務(wù)實例的 delay 方法。
ProcessPodcast::dispatch($podcast)
->delay(Carbon::now()->addMinutes(10));
通過推送任務(wù)到不同的隊列，可以給隊列任務(wù)分類，甚至可以控制給不同的隊列分配多少任務(wù)。要指定隊列的話，就調(diào)用任務(wù)實例的 onQueue 方法：
ProcessPodcast::dispatch($podcast)->onQueue("processing");
如果使用了多個隊列連接，可以將任務(wù)推到指定連接。要指定連接的話，可以在分發(fā)任務(wù)的時候使用 onConnection 方法：
ProcessPodcast::dispatch($podcast)->onConnection("redis
");
參考資源
KingFer
文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/29320.html
摘要：把因執(zhí)行超時的隊列從集合重新到當前執(zhí)行的隊列中。從要執(zhí)行的隊列中取任務(wù)可以看到在取要執(zhí)行的隊列的時候，同時會放一份到一個有序集合中，并使用過期時間戳作為分值。 （原文鏈接：https://blog.tanteng.me/2017/...） 在 Laravel 中使用 Redis 處理隊列任務(wù)，框架提供的功能非常強大，但是最近遇到一個問題，就是發(fā)現(xiàn)一個任務(wù)被多次執(zhí)行，這是為什么呢？ 先說...
摘要：在使用中的隊列時，產(chǎn)生沖突干擾。文件中的配置部分至此，兩個項目的隊列沖突原因就找到了。隊列監(jiān)聽最后遇到問題，莫要病急亂投醫(yī)。從代碼入手，分析理解實現(xiàn)原理，找對點，解決方法也許很簡單，。 問題 公司項目使用Laravel的開發(fā)的兩個項目在同一個測試服務(wù)器部署，公用同一個redis。在使用laravel中的隊列時，產(chǎn)生沖突干擾。 查找問題原因 在laravel 隊列的操作類 Illumin...
摘要：已經(jīng)取消了參數(shù)，都用來執(zhí)行。取數(shù)據(jù)的過程事物處理已經(jīng)打開。取得符合條件的隊列后程序會更新該條數(shù)據(jù)，并且更新完后即。 connections => [ .... database => [ driver => database, table => jobs, queue => defaul...
摘要：隊列的目的是將耗時的任務(wù)延時處理，比如發(fā)送郵件，從而大幅度縮短請求和相應(yīng)的時間。每一種隊列驅(qū)動的配置都可以在該文件中找到，包括數(shù)據(jù)庫，，，，以及同步本地使用驅(qū)動。處理完畢后當前任務(wù)會自動刪除?；揪拖旅孢@個樣到此隊列簡單配置與使用就結(jié)束了。 概述 什么是隊列？ 百度百科是這樣說的 隊列是在傳輸過程中保存數(shù)據(jù)的容器。 舉幾個生活中例子： iphone手機新款發(fā)布，三里屯iphone進的...
閱讀 1372·2021-11-24 09:39
閱讀 1361·2021-11-04 16:12
閱讀 2705·2021-09-24 09:47
閱讀 3350·2021-09-01 10:50
閱讀 1490·2019-08-30 15:55
閱讀 1439·2019-08-30 15:43
閱讀 656·2019-08-30 11:08
閱讀 3591·2019-08-23 18:33
