PHP Laravel 隊列技巧：Fail、Retry 或者 Delay

来源:https://www.cnblogs.com/a609251438/archive/2019/11/10/11831867.html
-Advertisement-
Play Games

當創建隊列jobs、監聽器或訂閱伺服器以推送到隊列中時，您可能會開始認為，一旦分派，隊列工作器決定如何處理您的邏輯就完全由您自己決定了。 嗯……並不是說你不能從作業內部與隊列工作器交互，但是通常情況下，哪怕你做了，也是沒必要的。 這個神奇的騷操作的出現是因為“InteractsWithQueue”這 ...


當創建隊列jobs、監聽器或訂閱伺服器以推送到隊列中時，您可能會開始認為，一旦分派，隊列工作器決定如何處理您的邏輯就完全由您自己決定了。

嗯……並不是說你不能從作業內部與隊列工作器交互，但是通常情況下，哪怕你做了，也是沒必要的。

這個神奇的騷操作的出現是因為“InteractsWithQueue”這個trait。.當排隊作業正在從隊列中拉出, 這個 [CallQueuedListener](https://github.com/laravel/framework/blob/5.8/src/Illuminate/Events/CallQueuedListener.php#L90-L104) 會檢查它是否在使用 InteractsWithQueue trait, 如果是的話，框架會將底層的“隊列jobs”實例註入到內部。

這個 “任務” 實例類似於一個包裝了真正的 Job 類的驅動，其中包含隊列連接和嘗試等信息。

背景
我將以一個轉碼 Job 為例。 這是一個將廣播音頻文件轉換成192kbps MP3格式的任務。因為這是在自由轉碼隊列中設置的，所以它的作用有限。

 

檢查嘗試次數
attempts()是被調用的第一個方法, 顧名思義，它返回嘗試次數,一個隊列 job總是伴隨著一個attempt啟動。

此方法旨在與其他方法一起使用 ..., 類似 fail() 或者 release() (delay). 為了便於說明，我們將通知用戶第幾次重試: 每次我們嘗試在空閑隊列中轉換（轉換代碼）時，我們都會通知用戶我們正在第幾次重試，讓他可以選擇取消將來的轉換（轉換代碼）。

 1 <?php
 2 namespace App\Jobs;
 3 use App\Podcast;
 4 use Transcoder\Transcoder;
 5 use Illuminate\Bus\Queueable;
 6 use Illuminate\Queue\SerializesModels;
 7 use App\Notifications\PodcastTranscoded;
 8 use Illuminate\Queue\InteractsWithQueue;
 9 use Illuminate\Foundation\Bus\Dispatchable;
10 use App\Notifications\RetyingPodcastTranscode;
11 class TranscodePodcast
12 {
13 use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
14 /**
15 * Transcoder Instance
16 *
17 * @var \App\Podcast
18 */
19 protected $podcast;
20 /**
21 * 創建一個新的轉碼podcast實例。
22 *
23 * @param \App\Podcast $podcast
24 * @return void
25 */
26 public function __construct(Podcast $podcast)
27 {
28 $this->podcast = $podcast;
29 }
30 /**
31 * 執行隊列job.
32 *
33 * @param \Transcoder\Transcoder $podcast
34 * @return void
35 */
36 public function handle(Transcoder $transcoder)
37 {
38 // 告訴用戶我們第幾次重試
39 if ($this->attempts() > 1) {
40 $this->podcast->publisher->notify(new RetryingPodcastTranscode($this->podcast, $this->attempts());
41 }
42 $transcoded = $this->transcoder->setFile($event->podcast)
43 ->format('mp3')
44 ->bitrate(192)
45 ->start();
46 
47 
48 // 將轉碼podcast與原始podcast關聯
49 $this->podcast->transcode()->associate($transcoded);
50 
51 // 通知podcast的發佈者他的podcast已經準備好了
52 
53 $this->publisher->notify(new PodcastTranscoded($this->podcast));
54 }
55 }

 


告訴用戶我們將在第幾次時重試某些內容，這在邏輯預先失敗時很有用，讓用戶（或開發人員）檢查出了什麼問題，但當然您可以做更多的事情。

就我個人而言，我喜歡在“Job作業”失敗後再這樣做，如果還有重試時間，告訴他我們稍後會重試當然，這個例子只是為了舉例說明。

刪除作業隊列 Job
第二個方法就是 delete(). 跟你猜想的一樣，您可以從隊列中刪除當前的“隊列 Job”。 當隊列 Job或偵聽器由於多種原因排隊後不應處理時，這將會很方便，例如，考慮一下這個場景：在轉碼發生之前，上傳podcast的發佈者由於任何原因（比如TOS衝突）被停用，我們應該不處理podcast。

我們將在前面的示例中添加該代碼：

 1 <?php
 2 namespace App\Jobs;
 3 use App\Podcast;
 4 use Transcoder\Transcoder;
 5 use Illuminate\Bus\Queueable;
 6 use Illuminate\Queue\SerializesModels;
 7 use App\Notifications\PodcastTranscoded;
 8 use Illuminate\Queue\InteractsWithQueue;
 9 use Illuminate\Foundation\Bus\Dispatchable;
10 use App\Notifications\RetyingPodcastTranscode;
11 class TranscodePodcast
12 {
13 use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
14 /**
15 * Transcoder Instance
16 *
17 * @var \App\Podcast
18 */
19 protected $podcast;
20 /**
21 * 創建一個新的轉碼podcast實例。
22 *
23 * @param \App\Podcast $podcast
24 * @return void
25 */
26 public function __construct(Podcast $podcast)
27 {
28 $this->podcast = $podcast;
29 }
30 /**
31 * 執行隊列 job.
32 *
33 * @param \Transcoder\Transcoder $podcast
34 * @return void
35 */
36 public function handle(Transcoder $transcoder)
37 {
38 // 如果發佈伺服器已被停用，請刪除此隊列job
39 if ($this->podcast->publisher->isDeactivated()) {
40 $this->delete();
41 }
42 // 告訴用戶我們第幾次重試
43 if ($this->attempts() > 1) {
44 $this->podcast->publisher->notify(new RetryingPodcastTranscode($this->podcast, $this->attempts());
45 }
46 $transcoded = $this->transcoder->setFile($event->podcast)
47 ->format('mp3')
48 ->bitrate(192)
49 ->start();
50 
51 // 將轉碼podcast與原始podcast關聯
52 $this->podcast->transcode()->associate($transcoded);
53 
54 // 通知podcast的發佈者他的podcast已經準備好了
55 $this->publisher->notify(new PodcastTranscoded($this->podcast));
56 }
57 }

 


如果需要刪除可能已刪除的模型上的作業，則可能需要 設置 [$deleteWhenMissingModels](https://laravel.com/docs/5.8/queues#ignoring-missing-models) 為真 t避免處理不存在的東西。

失敗的隊列job
當您需要控制人為破壞邏輯時，這非常非常方便, 因為使用空的“return”語句會將“Job”標記為已成功完成。您可以強制使排隊的作業失敗，希望出現異常，允許處理程式在可能的情況下稍後重試。

這使您在作業失敗時可以更好地控制在任何情況下，也可以使用“failed（）”方法, 它允許你 失敗後執行任何清潔操縱, 比如通知用戶或者刪除一些東西。

在此示例中，如果由於任何原因（如 CDN 關閉時）無法從存儲中檢索podcast ，則作業將失敗，並引發自定義異常。

 1 <?php
 2 namespace App\Jobs;
 3 use App\Podcast;
 4 use Transcoder\Transcoder;
 5 use Illuminate\Bus\Queueable;
 6 use Illuminate\Queue\SerializesModels;
 7 use App\Exceptions\PodcastUnretrievable;
 8 use App\Notifications\PodcastTranscoded;
 9 use Illuminate\Queue\InteractsWithQueue;
10 use Illuminate\Foundation\Bus\Dispatchable;
11 use App\Notifications\RetyingPodcastTranscode;
12 class TranscodePodcast
13 {
14 use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
15 /**
16 * 轉碼器實例
17 *
18 * @var \App\Podcast
19 */
20 protected $podcast;
21 /**
22 * 創建一個新的轉碼Podcast實例。
23 *
24 * @param \App\Podcast $podcast
25 * @return void
26 */
27 public function __construct(Podcast $podcast)
28 {
29 $this->podcast = $podcast;
30 }
31 /**
32 * 執行隊列 job.
33 *
34 * @param \Transcoder\Transcoder $podcast
35 * @return void
36 */
37 public function handle(Transcoder $transcoder)
38 {
39 // 如果發佈伺服器已被停用，請刪除此隊列job
40 if ($this->podcast->publisher->isDeactivated()) {
41 $this->delete();
42 }
43 //如果podcast不能從storage存儲中檢索，我們就會失敗。
44 if ($this->podcast->fileDoesntExists()) {
45 $this->fail(new PodcastUnretrievable($this->podcast));
46 }
47 // 告訴用戶我們第幾次重試
48 if ($this->attempts() > 1) {
49 $this->podcast->publisher->notify(new RetryingPodcastTranscode($this->podcast, $this->attempts());
50 }
51 
52 $transcoded = $this->transcoder->setFile($event->podcast)
53 ->format('mp3')
54 ->bitrate(192)
55 ->start();
56 
57 // 將轉碼podcast與原始podcast關聯
58 $this->podcast->transcode()->associate($transcoded);
59 
60 // 通知podcast的發佈者他的podcast已經準備好了
61 $this->publisher->notify(new PodcastTranscoded($this->podcast));
62 }
63 }

 


現在，進入最後的方法。

釋放（延遲）隊列job
這可能是trait性狀的最有用的方法, 因為它可以讓你在未來進一步推動這項隊列job. 此方法用於 隊列job速率限制.

除了速率限制之外，您還可以在某些不可用但希望在不久的將來使用它的情況下使用它同時，避免先發制人地失敗。

在最後一個示例中，我們將延遲轉碼以備稍後使用：如果轉碼器正在大量使用，我們將延遲轉碼5分鐘直到負載降低。

 1 <?php
 2 namespace App\Jobs;
 3 use App\Podcast;
 4 use Transcoder\Transcoder;
 5 use Illuminate\Bus\Queueable;
 6 use Illuminate\Queue\SerializesModels;
 7 use App\Exceptions\PodcastUnretrievable;
 8 use App\Notifications\PodcastTranscoded;
 9 use Illuminate\Queue\InteractsWithQueue;
10 use App\Notifications\TranscoderHighUsage;
11 use Illuminate\Foundation\Bus\Dispatchable;
12 use App\Notifications\RetyingPodcastTranscode;
13 class TranscodePodcast
14 {
15 use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
16 /**
17 * Transcoder Instance
18 *
19 * @var \App\Podcast
20 */
21 protected $podcast;
22 /**
23 * 創建一個新的轉碼podcast實例。
24 *
25 * @param \App\Podcast $podcast
26 * @return void
27 */
28 public function __construct(Podcast $podcast)
29 {
30 $this->podcast = $podcast;
31 }
32 /**
33 * 執行隊列job.
34 *
35 * @param \Transcoder\Transcoder $podcast
36 * @return void
37 */
38 public function handle(Transcoder $transcoder)
39 {
40 // 如果發佈伺服器已被停用，請刪除此隊列job
41 if ($this->podcast->publisher->isDeactivated()) {
42 $this->delete();
43 }
44 // 如果podcast不能從storage存儲中檢索，我們就會失敗。
45 if ($this->podcast->fileDoesntExists()) {
46 $this->fail(new PodcastUnretrievable($this->podcast));
47 }
48 
49 // 如果轉碼器使用率很高,我們將
50 // t延遲轉碼5分鐘. 否則我們可能會有拖延轉碼器進程的危險 
51 // 它會把所有的轉碼子進程都記錄下來。
52 if ($transcoder->getLoad()->isHigh()) {
53 $delay = 60 * 5;
54 $this->podcast->publisher->notify(new TranscoderHighUsage($this->podcast, $delay));
55 $this->release($delay);
56 }
57 // 告訴用戶我們第幾次重試
58 if ($this->attempts() > 1) {
59 $this->podcast->publisher->notify(new RetryingPodcastTranscode($this->podcast, $this->attempts());
60 }
61 
62 $transcoded = $this->transcoder->setFile($event->podcast)
63 ->format('mp3')
64 ->bitrate(192)
65 ->start();
66 
67 // 將轉碼podcast與原始podcast關聯
68 $this->podcast->transcode()->associate($transcoded);
69 
70 // 通知podcast的發佈者他的podcast已經準備好了
71 $this->publisher->notify(new PodcastTranscoded($this->podcast));
72 }
73 }

 


我們可以使用一些特殊方法，例如，獲得分配給轉碼器的一些時隙，如果轉碼器時隙已滿，則延遲作業。 在排隊的工作中，你能做的就只有這些了。排隊愉快。


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 深度解密Go語言之 pprof
    過早的優化是萬惡之源。而在真正遇到瓶頸的時候，pprof 可以快速定位到需要優化的地方。 ...
  • java 垃圾回收機制
    垃圾收集GC（Garbage Collection）是Java語言的核心技術之一， 在Java中，程式員不需要去關心記憶體動態分配和垃圾回收的問題，這一切都交給了JVM來處理。 一. jvm的記憶體結構 垃圾回收都是基於記憶體去回收的，因此，先要對記憶體結構有一個大概的瞭解 Java記憶體運行時區域大概分了三 ...
  • 海量數據展示(二）-性能優化
    背景 之前做的海量數據數據展示，在預處理速度和線上渲染上還有有所欠缺，本文中進行一些優化工作,使得九分鐘處理完一千多萬面數據的3 12級矢量切片，線上瀏覽數據請求時間控制在10s左右。 準備 軟體環境:PostGIS(3.0.0rc2 r17909)和 PostgreSQL( 12.0, compi ...
  • Tomcat啟動找不到項目依賴jar的解決方式
    Eclipse部署多模塊項目到tomcat，啟動時找不到jar的解決方法。 ...
  • PHP 面試官問：你說說Redis的幾個過期策略？
    在使用redis時，一般會設置一個過期時間，當然也有不設置過期時間的，也就是永久不過期。當設置了過期時間，redis是如何判斷是否過期，以及根據什麼策略來進行刪除的。 設置過期時間　expire key time(以秒為單位) 這是最常用的方式setex(String key, int second ...
  • linux scp命令 將數據從一臺linux伺服器複製到另一臺linux伺服器
    scp是secure copy的簡寫，用於在Linux下進行遠程拷貝文件的命令，和它類似的命令有cp，不過cp只是在本機進行拷貝不能跨伺服器，而且scp傳輸是加密的。可能會稍微影響一下速度。當你伺服器硬碟變為只讀 read only system時，用scp可以幫你把文件移出來。另外，scp還非常不 ...
  • PHP Laravel-包含你自己的幫助函數
    你可能想創建一個在應用的任何地方都可以訪問的函數，這個教程將幫你實現 👏 很多教程都會說，你在 composer.json 這個文件中通過添加一個自動載入的文件，就可以實現這個需求。但我認為這不是一個好的方式，當你在 helpers.php 文件中添加了更多的函數時，可讀性將變得很差。 下麵我將介 ...
  • RabbitMQ-交換機模式
    在說正題之前先解釋一下交換機模式是個籠統的稱呼，它不是一個單獨的模式（包括了訂閱模式，路由模式和主題模式），交換機模式是一個比較常用的模式，主要是為了實現數據的同步。 首先，說一下訂閱模式，就和字面上的意思差不多主要就是一個生產者，多個消費者，同一個消息被多個消費者獲取，先看一下官網的圖示 整體執行 ...
一周排行
    -Advertisement-
    Play Games
  • C#自增運算符詳解：++i與i++的區別及應用場景
    概述：在C#中，++i和i++都是自增運算符，其中++i先增加值再返回，而i++先返回值再增加。應用場景根據需求選擇，首碼適合先增後用，尾碼適合先用後增。詳細示例提供清晰的代碼演示這兩者的操作時機和實際應用。 在C#中，++i 和 i++ 都是自增運算符，但它們在操作上有細微的差異，主要體現在操作的 ...
  • Taurus.MVC 性能壓力測試（ap 壓測 和 linux 下wrk 壓測）：.NET 版本
    上次發佈了：Taurus.MVC 性能壓力測試（ap 壓測 和 linux 下wrk 壓測）：.NET Core 版本，今天計劃準備壓測一下 .NET 版本，來測試並記錄一下 Taurus.MVC 框架在 .NET 版本的性能，以便後續持續優化改進。 為了方便對比，本文章的電腦環境和測試思路，儘量和... ...
  • 一文讀懂.NET WebAPI中FromRoute、FromQuery、FromBody的關鍵角色與用法
    .NET WebAPI作為一種構建RESTful服務的強大工具，為開發者提供了便捷的方式來定義、處理HTTP請求並返迴響應。在設計API介面時，正確地接收和解析客戶端發送的數據至關重要。.NET WebAPI提供了一系列特性，如[FromRoute]、[FromQuery]和[FromBody]，用 ...
  • WPF/C#實現圖像濾鏡優化方案：打造炫目視覺體驗！
    原因：我之所以想做這個項目，是因為在之前查找關於C#/WPF相關資料時，我發現講解圖像濾鏡的資源非常稀缺。此外，我註意到許多現有的開源庫主要基於CPU進行圖像渲染。這種方式在處理大量圖像時，會導致CPU的渲染負擔過重。因此，我將在下文中介紹如何通過GPU渲染來有效實現圖像的各種濾鏡效果。 生成的效果 ...
  • Bogus 實戰：使用 Bogus 和 EFCore 生成模擬數據和種子數據【完整教程】
    引言 上一章我們介紹了在xUnit單元測試中用xUnit.DependencyInject來使用依賴註入，上一章我們的Sample.Repository倉儲層有一個批量註入的介面沒有做單元測試，今天用這個示例來演示一下如何用Bogus創建模擬數據 ，和 EFCore 的種子數據生成 Bogus 的優 ...
  • WPF隨筆收錄-實時繪製心率曲線
    一、前言 在自己的項目中，涉及到實時心率曲線的繪製，項目上的曲線繪製，一般很難找到能直接用的第三方庫，而且有些還是定製化的功能，所以還是自己繪製比較方便。很多人一聽到自己畫就害怕，感覺很難，今天就分享一個完整的實時心率數據繪製心率曲線圖的例子；之前的博客也分享給DrawingVisual繪製曲線的方 ...
  • wpf程式修改啟動入口註意事項
    如果你在自定義的 Main 方法中直接使用 App 類並啟動應用程式，但發現 App.xaml 中定義的資源沒有被正確載入，那麼問題可能在於如何正確配置 App.xaml 與你的 App 類的交互。 確保 App.xaml 文件中的 x:Class 屬性正確指向你的 App 類。這樣，當你創建 Ap ...
  • 記一次 .NET某爐膛鍋爐檢測系統 崩潰分析
    一：背景 1. 講故事 上個月有個朋友在微信上找到我，說他們的軟體在客戶那邊隔幾天就要崩潰一次，一直都沒有找到原因，讓我幫忙看下怎麼回事，確實工控類的軟體環境複雜難搞，朋友手上有一個崩潰的dump，剛好丟給我來分析一下。 二：WinDbg分析 1. 程式為什麼會崩潰 windbg 有一個厲害之處在於 ...
  • 一個.NET內置依賴註入的小型強化版
    前言 .NET生態中有許多依賴註入容器。在大多數情況下，微軟提供的內置容器在易用性和性能方面都非常優秀。外加ASP.NET Core預設使用內置容器，使用很方便。 但是筆者在使用中一直有一個頭疼的問題：服務工廠無法提供請求的服務類型相關的信息。這在一般情況下並沒有影響，但是內置容器支持註冊開放泛型服 ...
  • WPF隨筆收錄-DataGrid固定右側列
    一、前言 在項目開發過程中，DataGrid是經常使用到的一個數據展示控制項，而通常表格的最後一列是作為操作列存在，比如會有編輯、刪除等功能按鈕。但WPF的原始DataGrid中，預設只支持固定左側列，這跟大家習慣性操作列放最後不符，今天就來介紹一種簡單的方式實現固定右側列。（這裡的實現方式參考的大佬 ...
所有分類