When you create a queue job, listener, or subscriber to push to the queue, you may begin to think that once dispatched, it's up to you, the queue worker, to decide what to do with your logic.
Um... It's not that you can't interact with queue workers from within a job, but usually, even if you do, it's unnecessary.
This magical operation is due to the trail of "interactions with queue". . when the queued job is being pulled out of the queue, the [callqueuedlistener] (https://github.com/laravel/framework/blob/5.8/src/illuminate/events/callqueuedlistener.php ා l90-l104) will check whether it is using the interactionwithqueue trail. If so, the framework will inject the underlying "queue jobs" instance into the interior.
This "task" instance is similar to a driver that wraps the real Job class and contains information such as queue connections and attempts.
background
I'll take a transcoding Job as an example This is a task of converting broadcast audio files to 192kbps MP3 format. Because this is set in the free transcoding queue, its function is limited.
Number of check attempts
attempts() is the first method to be called. As the name implies, it returns the number of attempts. A queue job is always started with an attempt.
This method is designed to be used with other methods, such as fail() or release() (delay). For the sake of explanation, we will inform the user of the retries: every time we try to convert (transcoding) in the idle queue, we will inform the user that we are trying the retries, so that he can choose to cancel the future conversion (transcoding).
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 * Create a new transcoding podcast instance. 22 * 23 * @param \App\Podcast $podcast 24 * @return void 25 */ 26 public function __construct(Podcast $podcast) 27 { 28 $this->podcast = $podcast; 29 } 30 /** 31 * Execute queue job 32 * 33 * @param \Transcoder\Transcoder $podcast 34 * @return void 35 */ 36 public function handle(Transcoder $transcoder) 37 { 38 // Tell the user how many times we try again 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 // Transcoding podcast And primitive podcast Relation 49 $this->podcast->transcode()->associate($transcoded); 50 51 // notice podcast The publisher of podcast It's ready 52 53 $this->publisher->notify(new PodcastTranscoded($this->podcast)); 54 } 55 }
Tell the user that we will try something again on the next few times, which is useful when the logic fails in advance, and let the user (or developer) check what's wrong, but of course you can do more.
Personally, I like to do this after the "Job job" fails. If there is still time to retry, tell him that we will try again later. Of course, this example is just for illustration.
Delete Job queue Job
The second method is delete(). As you guessed, you can delete the current "queue Job" from the queue When a queue Job or listener should not be processed after queuing for a variety of reasons, it will be convenient. For example, consider this scenario: before transcoding, the publisher who uploads the podcast is disabled for any reason (such as TOS conflict), and we should not process the podcast.
We will add this code in the previous example:
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 * Create a new transcoding podcast instance. 22 * 23 * @param \App\Podcast $podcast 24 * @return void 25 */ 26 public function __construct(Podcast $podcast) 27 { 28 $this->podcast = $podcast; 29 } 30 /** 31 * Execute queue job 32 * 33 * @param \Transcoder\Transcoder $podcast 34 * @return void 35 */ 36 public function handle(Transcoder $transcoder) 37 { 38 // If the publisher has been deactivated, delete this queue job 39 if ($this->podcast->publisher->isDeactivated()) { 40 $this->delete(); 41 } 42 // Tell the user how many times we try again 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 // Transcoding podcast And primitive podcast Relation 52 $this->podcast->transcode()->associate($transcoded); 53 54 // notice podcast The publisher of podcast It's ready 55 $this->publisher->notify(new PodcastTranscoded($this->podcast)); 56 } 57 }
If you need to delete jobs on a model that may have been deleted, you may need to set [$deletewhenmissingmodels] (https://laravel.com/docs/5.8/queues ා ignoring missing models) to true t to avoid processing nonexistent things.
Failed queue job s
This is very convenient when you need to control human destruction logic, because using an empty "return" statement marks "Job" as completed successfully. You can force queued jobs to fail, expect exceptions, and allow the handler to retry later if possible.
This gives you better control over any situation when a job fails, or you can use the "failed()" method, which allows you to perform any cleaning operation after a failure, such as notifying the user or deleting something.
In this example, if the podcast cannot be retrieved from storage for any reason, such as when the CDN is down, the job fails and a custom exception is thrown.
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 * Transcoder instance 17 * 18 * @var \App\Podcast 19 */ 20 protected $podcast; 21 /** 22 * Create a new transcoding Podcast instance. 23 * 24 * @param \App\Podcast $podcast 25 * @return void 26 */ 27 public function __construct(Podcast $podcast) 28 { 29 $this->podcast = $podcast; 30 } 31 /** 32 * Execute queue job 33 * 34 * @param \Transcoder\Transcoder $podcast 35 * @return void 36 */ 37 public function handle(Transcoder $transcoder) 38 { 39 // If the publisher has been deactivated, delete this queue job 40 if ($this->podcast->publisher->isDeactivated()) { 41 $this->delete(); 42 } 43 //If podcast Cannot from storage Retrieval in storage, we will fail. 44 if ($this->podcast->fileDoesntExists()) { 45 $this->fail(new PodcastUnretrievable($this->podcast)); 46 } 47 // Tell the user how many times we try again 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 // Transcoding podcast And primitive podcast Relation 58 $this->podcast->transcode()->associate($transcoded); 59 60 // notice podcast The publisher of podcast It's ready 61 $this->publisher->notify(new PodcastTranscoded($this->podcast)); 62 } 63 }
Now, go to the last method.
Release (delay) queue job s
This may be the most useful method for the trait of trait, because it allows you to further promote this queue job in the future. This method is used for queue job rate restriction
In addition to rate limiting, you can use it in situations where it is not available but you want to use it in the near future while avoiding preemptive failure.
In the last example, we'll delay transcoding for later use: if the transcoder is in heavy use, we'll delay transcoding for five minutes until the load drops.
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 * Create a new transcoding podcast instance. 24 * 25 * @param \App\Podcast $podcast 26 * @return void 27 */ 28 public function __construct(Podcast $podcast) 29 { 30 $this->podcast = $podcast; 31 } 32 /** 33 * Execute queue job 34 * 35 * @param \Transcoder\Transcoder $podcast 36 * @return void 37 */ 38 public function handle(Transcoder $transcoder) 39 { 40 // If the publisher has been deactivated, delete this queue job 41 if ($this->podcast->publisher->isDeactivated()) { 42 $this->delete(); 43 } 44 // If podcast Cannot from storage Retrieval in storage, we will fail. 45 if ($this->podcast->fileDoesntExists()) { 46 $this->fail(new PodcastUnretrievable($this->podcast)); 47 } 48 49 // If transcoder usage is high,We will 50 // Tdelay transcoding for 5 minutes. Otherwise, we may be in danger of delaying the transcoder process 51 // It will record all transcoding sub processes. 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 // Tell the user how many times we try again 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 // Transcoding podcast And primitive podcast Relation 68 $this->podcast->transcode()->associate($transcoded); 69 70 // notice podcast The publisher of podcast It's ready 71 $this->publisher->notify(new PodcastTranscoded($this->podcast)); 72 } 73 }
We can use some special methods, for example, to obtain some time slots assigned to the transcoder, and delay the job if the transcoder time slot is full That's all you can do in line work. Have a good time in line.