Queues trong Laravel

Để phục vụ cho dự án sắp tới thì mình sẽ dành time tìm hiểu 1 chút về Queue trong Laravel. Tài liệu này mình dịch từ Doc hướng dẫn của Laravel 5.0. Các bạn có thê xem bản tiếng Anh tại đây .

Configuration

Đây là 1 component trong Laravel cung cấp 1 API hỗ trợ xử lý các dịch vụ khác nhau có liên quan đến queue. Queue cho phép bạn có thể delay lại time xử lý 1 task nào đến 1 thời điểm nhất định, việc làm này sẽ giúp cho hệ thống/ ứng dụng của bạn chạy nhanh hơn và hiệu quả hơn rất nhiều.

File cấu hình queue được lưu trong config/queue.php. Trong file này bạn sẽ tìm thấy các cấu hình kết nói cho mỗi loại queue driver bao gồm database, Beanstalkd, IronMQ, Amazon SQS, Redis, null, và driver đồng bộ hóa cho local user. Driver null queue đơn thuần sẽ loại bỏ các queue jobs khiến chúng ko chạy nữa.

** Queue Database Table **

Để sử dụng database queue driver, bạn sẽ cần 1 table DB để lưu dữ các jobs. Để có được table này, chạy lênh Artisan queue:table để sinh ra migration tạo table.

php artisan queue:table

** Other Queue Dependencies **

Dưới đây la 1 số dependencies khác cho queue drivers :

Amazon SQS: aws/aws-sdk-php

Beanstalkd: pda/pheanstalk ~3.0

IronMQ: iron-io/iron_mq ~1.5

Redis: predis/predis ~1.0

Basic Usage

** Đầy 1 Job vào trong Queue **

Tất cả các jobs mà có thể queue được trong ứng dụng của bạn sẽ đc lưu trong App\Commands. Bạn có thể sẽ phải tạo 1 command queued mới sử dụng Artisan CLI:

php artisan make:command SendEmail --queued

Để push 1 job mới vào trong queue thì sử dụng phương thức Queue::push:

Queue::push(new SendEmail($message));

Chú ý: Trong ví dụ ở trên, chúng ta sử dụng Queue facade trực tiếp. Tuy nhiên, bạn có thể ra lệnh cho queued command thông qua Command Bus. Trong bài viết này chúng tôi sẽ sư dụng Queue facade, cách sử dụng thông qua Command Bus sẽ tương tự với các viết này.

Mặc định thì lệnh Artisan make:command sẽ sinh ra 1 lệnh "tự xử lý", có thể hiểu là 1 phương thức xử ly được add thêm vào lệnh. Phương thức này sẽ được gọi khi job được thực thi trong queue. Bạn có thể gợi ý bất kì 1 dependencies bạn cần trong phương thức xử lý và service container sẽ tự động liên keets nó :

public function handle(UserRepository $users)
{
    //
}

Nếu bạn muốn command của bạn có nhiều class handler rời rạc nhau, bạn nên add thêm --handler khi gọi lệnh make:command:

php artisan make:command SendEmail --queued --handler

Những handler được sinh ra sẽ được đặt trong App\Handlers\Commands và sẽ được xử lý ở IoC container

** Specifying The Queue / Tube For A Job **

Bạn có thể cần chỉ định 1 queue/tube job được gửi đi :

Queue::pushOn('emails', new SendEmail($message));

** Passing The Same Payload To Multiple Jobs **

Nếu cần truyền dữ liệu vào 1 vài queue jobs, bạn nên sử dụng phương thức Queue::bulk:

Queue::bulk([new SendEmail($message), new AnotherCommand]);

** Delaying The Execution Of A Job **

Thỉnh thoảng bạn cần delay xử lý của 1 queued job. Ví dụ, bạn muốn tạo queue của 1 job gửi mail cho khách hàng sau 15 phút sign-up. Bạn nên sử dụng phương thức Queue::later :

$date = Carbon::now()->addMinutes(15);

Queue::later($date, new SendEmail($message));

Trong ví dụ trên, chúng tôi sử dụng thư viện ngày Carbon để chỉ định việc delay.

Chú ý : Amazon SQS service sẽ có delay limit 900 giây (15 phút).

** Queues And Eloquent Models **

Nếu queued job của bạn cho phép model Eloquent trong cấu trúc của nó, thì chỉ có mỗi identifier cho model được serialized trong queue. Khi job được thực thi, hệ thống queue sẽ tự động nhận lại tất cả instance của model từ DB. Cách làm này rất rõ ràng đối với ứng dụng của bạn và hơn thế nữa sẽ ngăn ngừa các vấn đề có thể phát sinh từ việc serialize toàn bộ các instance của full Eloquent model.

** Deleting A Processed Job **

Một khi thực hiện 1 job, nó phải được xóa khỏi queue. Nêu ko có exception throw trong quá trình thực thi job đó thì việc xóa sẽ được thực hiển tự động. Nếu bạn muốn delete hoặc release 1 job thủ công, Illuminate\Queue\InteractsWithQueue cung cấp cho bạn phương thức delete or release để làm điều đó. Phương thức release chấp nhận 1 giá trị : số lượng giây bạn muốn chờ cho đến đến job được tái hiện lại.

public function handle(SendEmail $command)
{
    if (true)
    {
        $this->release(30);
    }
}

** Releasing A Job Back Onto The Queue **

Nếu 1 exception bị throw trong khi job đang được xử lý, nó sẽ tự động release back về queue cho nên nó có thể sẽ được thực hiện lại. Job sẽ tiếp tục bị release back đến lúc đạt được số lần attempt lớn nhất được cho phép trong ứng dụng của bạn. Số attempt lớn nhất sẽ được định nghĩa trong --tries sử dụng lệnh queue:listen hoặc là queue:work

** Checking The Number Of Run Attempts **

Nếu 1 exception xảy ra trong khi đang thực thi 1 job, nó sẽ tự động bị release back lại queue. Bạn có thể check số lần attempts cho phép chạy lại job sử dụng phương thức attempts :

if ($this->attempts() > 3)
{
    //
}

Chú ý: Lênh/handlder của bạn phải sử dụng Illuminate\Queue\InteractsWithQueue để gọi phương thức này.

Queueing Closures

Bạn có thể cần push 1 Closure vào trong queue. Việc này rất thuận tiên cho những task đơn giản và nhanh cân được queue:

** Pushing A Closure Onto The Queue **

Queue::push(function($job) use ($id)
{
    Account::delete($id);

    $job->delete();
});

Khi sử dụng Iron.io push queue, bạn nên chú ý hơn đến việc queue Closure. Điêm kết thúc mà nhận được queue messages của bạn nên check cho 1 token để verify request có thật là từ Iron.io hay không. Ví dụ, push queue end-point của bạn nên như thế này:https://yourapp.com/queue/receive?token=SecretToken Bạn cũng có thể check giá trị của secret token trong ứng dụng của bạn trước khi marshal request queue.

Running The Queue Listener

Laravel có 1 task Artisan cho phép chạy jobs mới ngay khi nó được push vào queue. Bạn có thể chạy task này sử dụng lệnh queue:listen:

** Starting The Queue Listener **

php artisan queue:listen

Bạn cũng có thể chị định queue connection nào mà listener nên sử dụng:

php artisan queue:listen connection

Chú ý rằng 1 khi task được bắt đầu, nó sẽ chạy cho đến khi nó được dừng bằng tay. Bạn có thể sử dụng monitor như Supervisor để chắc chắn rằng queue listener ko dừng chạy. Bạn cũng có thể truyền comma-delimited để set tính ưu tiên trong viêc lắng nghe các queue:

php artisan queue:listen --queue=high,low

Trong ví dụ này, jobs trong high-connection sẽ luôn luôn được thực hiện trước khi chuyển đến jobs trong low-connection.

** Specifying The Job Timeout Parameter **

Bạn có thể chỉ định độ dài time (giây) cho phép mỗi jobs được chạy trong bao lâu:

php artisan queue:listen --timeout=60

** Specifying Queue Sleep Duration **

Thêm nữa, bạn co thể chị định số giây chờ trước khi kéo thêm job mới vào:

php artisan queue:listen --sleep=5

Chú ý rằng queue chỉ "sleeps" nếu ko có jobs nào trong queue. Nếu có thêm jobs đang avaible, queue sẽ tiếp tục thực hiện mà ko "sleeping".

** Processing The First Job On The Queue **

Để chỉ chạy job đầu tiên trong queue, bạn sử dụng queue:work:

php artisan queue:work

Daemon Queue Worker

queue:work bao gồm option --daemon cho việc ép các queue worker tiếp tục thực hiện jobs mà ko cần phải re-boot lại framework. Việc này sẽ làm giảm thiểu 1 lượng sử dụng đáng kế CPU nếu so sánh với việc queue:listen, nhưng cũng tăng thêm tính phức tạp trong quá trình xử lý.

Để chạy 1 queue woker trong deamon mode, sử dụng --daemon:

php artisan queue:work connection --daemon

php artisan queue:work connection --daemon --sleep=3

php artisan queue:work connection --daemon --sleep=3 --tries=3

Bạn có thể thấy, queue:work hỗ trợ hầu hết các options tương tự với queue:listen. Bạn cũng có thể sử dụng php artisan help queue:work để xem tất cả các options.

** Deploying With Daemon Queue Workers **

Cách đơn giản nhất để deploy ứng dụng sử dụng daemon queue worker là đặt ứng dụng vào mode maintenance tại thời điểm đấu của deploy. Việc này có thể làm được bằng cách sử dụng lệnh php artisan down. Một khi ứng dụng ở mode maintenance, Laravel sẽ ko chấp nhận bất kì 1 jobs mới ra khỏi trong queue, và tiếp tục tiến hành những jobs hiện tại.

Cách dễ nhất để khởi động your workers là chạy script:

php artisan queue:restart

Lệnh này sẽ ra lệnh tất cả các queue workers restart sau khi chúng kết thúc jobs hiện tại.

** Coding For Daemon Queue Workers **

Daemon queue wokers ko khởi động lại framework trước khi xử lý từng job. Do đó, bạn nên cẩn thận giải phóng bất kì resource nào nặng trước khi job được kết thúc. Ví dụ, nếu bạn đang thao tác xử lý ảnh với thư viện GD, bạn nên giải phóng bộ nhớ với imagedestroy khi bạn xong.

Tương tự như vây, kết nối DB của bạn có thể bị disconnect khi sử dụng long-running daemon. Bạn nên sử dụng DB::reconnect để chắc chắn là bạn có 1 connection mới.

Push Queues

Push queue cho phép bạn sử dụng Laravel 5 queue facilities mà ko cần chạy bất kì deamons hay listeners nào phía sau. Hiện nay, push queues chỉ được support bởi driver Iron.io. Trước khi bắt đầu, hãy tạo 1 account Iron.io và add account đấy vào config/queue.php.

** Registering A Push Queue Subscriber **

Tiếp, bạn có thể sử dụng queue:subscribe đẻ đăng kí 1 URL end-point co thể nhận pushed queue jobs.

php artisan queue:subscribe queue_name queue/receive

php artisan queue:subscribe queue_name http://foo.com/queue/receive

Bây giờ, khi bạn login vào Iron dashboard, bạn sẽ nhận được push queue mới của bạn, và subscribed URL. Bạn có thể subscribe bao nhiêu URLs mà bạn muốn đưa vào queue. Tiếp, bạn tạo 1 route cho queue/receive end-point và trả về response từ phương thức Queue::marshal:

Route::post('queue/receive', function()
{
    return Queue::marshal();
});

Phương thức marshal sẽ kiếm soát việc kích hoạt những class handlder job. Để kích hoạt 1 jobs vào trong push queue, chỉ cần sử dụng Queue::push/

Failed Jobs

Thỉnh thoàng sẽ có 1 vài queued job bị fail vì những lý do ko ngờ tới. Laravel cung cấp cho chúng ta 1 cách rấ thuận tiện để chỉ định số lần 1 jobs được chạy lại khi nó fail. Nếu jobs vượt quá số lần này, nó sẽ được lưu vào trong table failed_jobs. Table này có thể được thay tên trong file config/queue.php.

Để tạo 1 migration cho table failed_jobs, bạn sử dụng lệnh queue:failed-table:

php artisan queue:failed-table

Bạn có thể chỉ định số lần maximum cho 1 jobs để được lắng nghe sử dụng --tries trong lệnh queue:listen :

php artisan queue:listen connection-name --tries=3

Nếu bạn muốn đăng kí 1 event lắng nghe khi có 1 queue job bị fail, bạn sủ dụng phương thức Queue::failing.

Queue::failing(function($connection, $job, $data)
{
    //
});

Bạn có thể định nghĩa 1 phương thức là failed trong class queue job, cho phép bạn thực thi 1 action cụ thể khi có lỗi xảy ra:

public function failed()
{
    // Called when the job is failing...
}

** Retrying Failed Jobs **

Để view tất cả các jobs bị fail, sử dụng lệnh queue:failed

php artisan queue:failed

Lệnh queue:failed sẽ liệt kê ra job ID, connection, queue, and failure time. Job ID có thể sử dụng để thực thi lại những job bị fail. Ví dụ, để retry lại 1 Jobs bị fail có ID là 5, bạn có thể làm như sau :

php artisan queue:retry 5

Để xóa tất cả các jobs bị fail, bạn có thể sử dụng lênh queue:forget :

php artisan queue:forget 5

Để xóa tất cả các jobs bị fail, bạn có thể sử dụng lênh queue:flush :

php artisan queue:flush