Queue giúp request Laravel trả về nhanh hơn, nhưng đưa công việc sang nền không tự động làm hệ thống đáng tin cậy. Trong production, worker có thể dừng giữa chừng, API bên ngoài có thể timeout và cùng một job có thể được xử lý lại. Một thiết kế tốt phải chủ động kiểm soát retry, timeout, tính idempotent, tài nguyên và khả năng quan sát.
Khi nào nên đưa công việc vào queue?
Queue phù hợp với tác vụ không cần hoàn tất trước khi trả response: gửi email, tạo báo cáo, xử lý ảnh, đồng bộ dữ liệu, phát webhook hoặc nhập file lớn. Những thao tác quyết định trực tiếp kết quả mà người dùng đang chờ, chẳng hạn xác thực quyền hay giữ chỗ tồn kho, thường vẫn cần xử lý đồng bộ hoặc một thiết kế nghiệp vụ riêng.
Queue là ranh giới xử lý bất đồng bộ, không phải nơi giấu mọi đoạn mã chậm.
Chọn Redis và cài Laravel Horizon
Horizon cung cấp dashboard và cấu hình worker cho Laravel Queue dùng Redis. Tài liệu Laravel hiện tại yêu cầu Horizon sử dụng kết nối Redis và lưu ý Horizon chưa tương thích với Redis Cluster. Cài đặt bằng các lệnh sau:
composer require laravel/horizon
php artisan horizon:install
php artisan migrateĐặt queue connection trong môi trường production:
QUEUE_CONNECTION=redis
REDIS_HOST=127.0.0.1
REDIS_PORT=6379Redis không nên được mở trực tiếp ra Internet. Hãy giới hạn mạng, bật xác thực khi phù hợp, theo dõi bộ nhớ và thiết lập chính sách persistence dựa trên mức độ chấp nhận mất job của hệ thống.
Tạo job có giới hạn rõ ràng
Một job production cần giới hạn số lần thử, thời gian chạy và khoảng nghỉ giữa các lần retry. Mảng backoff giúp tăng dần thời gian chờ khi dịch vụ phụ thuộc đang lỗi tạm thời.
<?php
namespace App\Jobs;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Throwable;
class SyncInvoice implements ShouldQueue
{
use Queueable;
public int $tries = 5;
public int $timeout = 90;
public bool $failOnTimeout = true;
public function __construct(public int $invoiceId) {}
public function backoff(): array
{
return [10, 30, 120, 300];
}
public function handle(): void
{
// Gọi service với connect timeout và request timeout riêng.
}
public function failed(?Throwable $exception): void
{
// Ghi nhận ngữ cảnh và phát cảnh báo, không nuốt lỗi.
}
}Chỉ retry lỗi tạm thời như timeout, 429 hoặc 5xx. Lỗi validation, thiếu dữ liệu bắt buộc hay lỗi quyền truy cập thường không tự biến mất sau lần thử tiếp theo; hãy cho job fail có chủ đích và đưa vào luồng xử lý phù hợp.
Ba mốc thời gian phải khớp nhau
Một cấu hình an toàn thường tuân theo thứ tự:
job timeout < Horizon supervisor timeout < Redis retry_afterVí dụ job timeout 90 giây, Horizon timeout 100 giây và retry_after 120 giây. Horizon cần có cơ hội dừng worker bị treo trước khi queue coi job là mất và giao lại. Nếu timeout của worker dài hơn hoặc quá sát retry_after, hai worker có thể cùng xử lý một job.
Ở tầng trình quản lý tiến trình, stopwaitsecs phải lớn hơn thời gian chạy dài nhất của job để deploy hoặc restart không giết tiến trình giữa công việc hợp lệ.
Dispatch sau khi transaction đã commit
Nếu job phụ thuộc vào dữ liệu vừa ghi trong transaction, worker nhanh có thể chạy trước khi transaction commit và không tìm thấy bản ghi. Có thể bật after_commit cho queue connection hoặc chỉ định trên lần dispatch:
DB::transaction(function () use ($order) {
$order->markAsPaid();
GenerateInvoice::dispatch($order->id)->afterCommit();
});Chỉ truyền ID và dữ liệu tối thiểu vào job. Laravel có thể serialize model thành định danh rồi tải lại khi job chạy; relationship đã load cũng có thể được tải lại ngoài phạm vi query ban đầu. Truyền ID giúp contract rõ hơn; dữ liệu tải lại không phải snapshot lúc dispatch.
Idempotency: phòng tuyến chống xử lý trùng
Queue thường mang ngữ nghĩa giao ít nhất một lần. Job có thể hoàn tất tác động bên ngoài nhưng worker dừng trước khi xác nhận, khiến job được giao lại. Vì vậy thao tác như trừ tiền, tạo hóa đơn hoặc gửi webhook phải chịu được việc gọi lặp.
- Dùng khóa nghiệp vụ duy nhất, ví dụ
invoice_id + action. - Tạo unique constraint trong database thay vì chỉ kiểm tra bằng mã ứng dụng.
- Lưu trạng thái và cập nhật nghiệp vụ trong cùng transaction khi có thể.
- Dùng transactional outbox nếu cần phối hợp database với message hoặc API bên ngoài.
- Có thể dùng
ShouldBeUniqueđể hạn chế enqueue trùng, nhưng không coi đó là bảo đảm “exactly once”.
Chia queue theo độ ưu tiên và tài nguyên
Không nên để email, ảnh nặng và tác vụ thanh toán cạnh tranh trong một queue. Một cách chia dễ vận hành là high, default, notifications và media. Trong config/horizon.php, tạo supervisor riêng cho workload cần tài nguyên hoặc độ ưu tiên khác nhau.
'production' => [
'supervisor-critical' => [
'connection' => 'redis',
'queue' => ['high'],
'balance' => 'simple',
'processes' => 4,
'tries' => 3,
'timeout' => 100,
],
'supervisor-default' => [
'connection' => 'redis',
'queue' => ['default', 'notifications'],
'balance' => 'auto',
'minProcesses' => 2,
'maxProcesses' => 10,
'balanceMaxShift' => 1,
'balanceCooldown' => 3,
'tries' => 5,
'timeout' => 100,
'backoff' => [10, 30, 120],
],
],auto phân bổ worker theo tải, nhưng không bảo đảm ưu tiên tuyệt đối giữa các queue. Workload quan trọng nên có supervisor riêng và giới hạn worker dựa trên CPU, RAM, kết nối database và rate limit của dịch vụ phụ thuộc.
Chạy Horizon như một dịch vụ
Production cần Supervisor, systemd hoặc trình quản lý tiến trình tương đương để tự khởi động lại Horizon khi máy reboot hoặc tiến trình lỗi. Tiến trình chính là:
php artisan horizonDashboard Horizon chứa thông tin vận hành, vì vậy phải giới hạn quyền truy cập qua gate trong HorizonServiceProvider, đồng thời đặt sau HTTPS và cơ chế xác thực của ứng dụng.
Deploy mà không để worker chạy mã cũ
Worker là tiến trình sống lâu và không tự nạp lại source code sau mỗi request. Với Laravel 13, có thể dùng php artisan reload trong quy trình deploy để reload các dịch vụ dài hạn. Riêng Horizon, php artisan horizon:terminate yêu cầu tiến trình hiện tại thoát nhẹ nhàng sau khi hoàn tất job; trình quản lý tiến trình sẽ khởi động phiên bản mới.
php artisan migrate --force
php artisan optimize
php artisan horizon:terminateKhông deploy thay đổi không tương thích với payload job đang chờ. Khi đổi constructor hoặc định dạng dữ liệu, hãy hỗ trợ payload cũ trong giai đoạn chuyển tiếp hoặc làm cạn queue trước khi phát hành.
Xử lý failed jobs và quan sát hệ thống
Đội vận hành cần biết job thất bại vì đâu trước khi retry hàng loạt:
php artisan queue:failed
php artisan queue:retry <job-id>
php artisan queue:retry all
php artisan horizon:forget <job-id>Dashboard và cảnh báo nên theo dõi độ trễ queue, tuổi của job lâu nhất, throughput, runtime p95/p99, số lần retry, failed jobs, worker restart, bộ nhớ Redis và lỗi từ dịch vụ phụ thuộc. Thiết lập snapshot Horizon theo scheduler và prune dữ liệu theo thời gian lưu giữ phù hợp.
Kiểm thử trước khi đưa lên production
- Dùng
Queue::fake()để xác minh job được dispatch đúng queue và đúng dữ liệu. - Kiểm thử
handle()riêng với lỗi tạm thời và lỗi vĩnh viễn. - Chạy integration test với Redis và worker thật cho đường đi quan trọng.
- Mô phỏng worker dừng sau tác động nghiệp vụ nhưng trước lúc ACK.
- Kiểm tra nhiều job cùng chạy trên một khóa nghiệp vụ.
- Thử deploy khi queue đang có tải và xác minh worker mới dùng đúng phiên bản mã.
Checklist production
- Redis chỉ truy cập từ mạng tin cậy và có giám sát bộ nhớ.
- Mọi job có timeout, số lần thử và backoff hữu hạn.
- Quan hệ timeout tuân theo
job < Horizon < retry_after. - Job phụ thuộc transaction chỉ dispatch sau commit.
- Tác động nghiệp vụ quan trọng có idempotency key và unique constraint.
- Queue nặng hoặc quan trọng có supervisor riêng.
- Horizon chạy dưới process monitor và dashboard được bảo vệ.
- Deploy có bước reload hoặc terminate worker nhẹ nhàng.
- Failed jobs có cảnh báo, người chịu trách nhiệm và quy trình replay.
Kết luận
Laravel Queue đáng tin cậy không nằm ở số lượng worker, mà ở các giới hạn được thiết kế rõ: job có thể retry nhưng không tạo tác động trùng, worker có thể restart mà không mất việc, và đội vận hành nhìn thấy queue đang chậm ở đâu. Redis và Horizon cung cấp nền tảng tốt; tính an toàn đến từ timeout hợp lý, idempotency, chia tải và quy trình deploy có kiểm soát.
Nguồn tham khảo
Bài 26 TaskFlow: chạy job trên database queue
Phần trên là hướng dẫn Redis/Horizon cho production. TaskFlow thực hành bằng database queue có sẵn để chạy được trên môi trường khóa học, chưa cài Horizon hoặc Redis. Không cần cài thêm dịch vụ chỉ để theo dõi đường đi dispatch → lưu job → worker → xử lý.
Khi CreateTask phát TaskCreated sau commit, listener QueueProjectCountInvalidation đưa job vào queue maintenance. Job xóa đúng key bộ đếm project của bài 19. Đây là ví dụ nhẹ phục vụ học queue; riêng thao tác cache forget thường không cần bất đồng bộ vì chi phí dispatch có thể lớn hơn công việc.
Job mang project ID, giới hạn tries=3, timeout=15 giây và backoff. Xóa lại cùng key không tạo dữ liệu trùng; không dùng Cache::flush. Count vẫn có TTL 30 giây làm đường dự phòng, nhưng invalidation không cung cấp nhất quán mạnh hoặc bảo đảm giao nhận. Reader đang tính có thể ghi lại giá trị cũ sau khi key bị xóa; process chết giữa commit và enqueue vẫn có thể mất job, vì chưa có outbox.
tests/Feature/TaskQueueTest.php
<?php
namespace Tests\Feature;
use App\Actions\CreateTask;
use App\Data\CreateTaskData;
use App\Jobs\InvalidateProjectTaskCount;
use App\Models\Project;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Tests\TestCase;
class TaskQueueTest extends TestCase
{
protected function setUp(): void
{
parent::setUp();
config(['database.default' => 'sqlite', 'database.connections.sqlite.database' => ':memory:',
'database.connections.sqlite.url' => null, 'cache.default' => 'array',
'queue.default' => 'database', 'queue.connections.database.connection' => 'sqlite']);
DB::purge('sqlite');
$this->artisan('migrate', ['--force' => true])->assertSuccessful();
}
public function test_after_commit_job_is_processed_by_database_worker(): void
{
$project = Project::factory()->create();
$key = 'taskflow:project:'.$project->id.':task-count:v1';
Cache::put($key, 0, 30);
DB::transaction(function () use ($project) {
app(CreateTask::class)->handle($project, new CreateTaskData('Queued task'));
$this->assertDatabaseCount('jobs', 0);
});
$this->assertDatabaseCount('jobs', 1);
$this->assertSame(0, Cache::get($key));
$this->artisan('queue:work', ['connection' => 'database', '--queue' => 'maintenance',
'--once' => true, '--tries' => 3, '--timeout' => 15])->assertSuccessful();
$this->assertDatabaseCount('jobs', 0);
$this->assertNull(Cache::get($key));
$this->assertDatabaseCount('failed_jobs', 0);
}
public function test_rollback_does_not_enqueue_a_job(): void
{
$project = Project::factory()->create();
try {
DB::transaction(function () use ($project) {
app(CreateTask::class)->handle($project, new CreateTaskData('Rollback'));
throw new \RuntimeException('Rollback');
});
} catch (\RuntimeException $exception) {
$this->assertSame('Rollback', $exception->getMessage());
}
$this->assertDatabaseCount('tasks', 0);
$this->assertDatabaseCount('jobs', 0);
}
public function test_duplicate_execution_only_invalidates_its_project(): void
{
Cache::put('taskflow:project:1:task-count:v1', 4, 30);
Cache::put('taskflow:project:2:task-count:v1', 8, 30);
$job = new InvalidateProjectTaskCount(1);
$job->handle();
$job->handle();
$this->assertNull(Cache::get('taskflow:project:1:task-count:v1'));
$this->assertSame(8, Cache::get('taskflow:project:2:task-count:v1'));
}
}app/Listeners/QueueProjectCountInvalidation.php
<?php
namespace App\Listeners;
use App\Events\TaskCreated;
use App\Jobs\InvalidateProjectTaskCount;
class QueueProjectCountInvalidation
{
public function handle(TaskCreated $event): void
{
InvalidateProjectTaskCount::dispatch($event->projectId)->onQueue('maintenance')->afterCommit();
}
}app/Jobs/InvalidateProjectTaskCount.php
<?php
namespace App\Jobs;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Support\Facades\Cache;
class InvalidateProjectTaskCount implements ShouldQueue
{
use Queueable;
public int $tries = 3;
public int $timeout = 15;
public function __construct(public int $projectId) {}
public function backoff(): array
{
return [5, 15, 30];
}
public function handle(): void
{
Cache::forget('taskflow:project:'.$this->projectId.':task-count:v1');
}
}Chạy trên môi trường phát triển
# Trong .env phát triển:
QUEUE_CONNECTION=database
# Sau khi xác nhận đúng database phát triển:
php artisan migrate
php artisan queue:work database --queue=maintenance --tries=3 --timeout=15
# Kiểm tra:
php artisan test --filter=TaskQueueTest
php artisan testGiữ worker ở terminal riêng rồi tạo task qua giao diện hoặc API. Worker chỉ nghe default sẽ không lấy job maintenance. Database connection có retry_after mặc định 90 giây, lớn hơn timeout job; timeout thực tế còn phụ thuộc khả năng process signal/PCNTL của môi trường. Không coi kiểm thử Windows này là bằng chứng worker treo chắc chắn bị dừng sau 15 giây.
Test dùng database queue thật trên SQLite bộ nhớ và chạy queue:work --once trong cùng tiến trình test; không dùng Queue::fake. Nó chứng minh job chưa có trước commit, có sau commit, worker lấy/xóa job và xóa cache; rollback không enqueue; handle lặp không ảnh hưởng key project khác. Suite đạt 79 test, 363 assertions. Chưa mô phỏng crash, backoff khi dependency lỗi, worker nhiều process hay Redis/Horizon.
Ở production, chọn cache chia sẻ phù hợp để worker và web nhìn cùng key; array cache riêng process không đạt yêu cầu đó. Giám sát queue maintenance, failed jobs và tuổi job; không retry all trước khi hiểu lỗi. Bài scheduler kế tiếp sẽ bổ sung tác vụ định kỳ, không thay worker queue.
Điều hướng: Bài 25 · Lộ trình.
Chưa có bình luận. Hãy là người đầu tiên chia sẻ ý kiến.