Bài viết này không phải là một tài liệu tham khảo khô khan, mà bắt đầu từ một vấn đề rất thật mà tôi gặp phải: số lượng message trong hàng đợi dâng lên rất cao, nhưng thời gian xử lý lại chậm một cách khó hiểu. Khi kiểm tra lại, tôi phát hiện số message đang inflight chỉ vỏn vẹn là 1, trong khi có tới 3 worker đang chạy trên cùng một EC2 instance. Từ dấu hỏi đó, tôi đã đi qua một hành trình khá thú vị: hiểu lại bản chất của SQS FIFO, tối ưu throughput bằng nhiều Message Group ID, rồi cuối cùng migrate toàn bộ sang serverless với ECS Fargate và tự động scale theo backlog.

Nếu bạn cũng đang chạy php artisan queue:work sqs trên một con EC2 nào đó và cảm giác nó "chậm một cách khó hiểu", hy vọng bài viết này sẽ giúp bạn tiết kiệm được vài buổi tối debug (có lẽ vậy).

Bối cảnh hệ thống

Hệ thống của tôi khá điển hình:

Laravel v12 làm backend. Hàng đợi dùng Amazon SQS FIFO (không phải Standard). Worker chạy bằng php artisan queue:work sqs, được giám sát bởi Supervisord với 3 processes trên một con EC2.

Lý do chọn FIFO thay vì Standard rất rõ ràng: có những tác vụ chỉ được phép xử lý đúng một lần, ví dụ gửi thông báo giao dịch cho người dùng. Gửi hai lần là một sự cố. FIFO cho tôi hai thứ mà Standard không có: thứ tự xử lý trong cùng một nhóm, và cơ chế khử trùng lặp (deduplication) giúp tiến gần tới "exactly once". Nhờ vậy tôi tạm thời chưa phải viết logic idempotency phức tạp trong tầng ứng dụng.

Mọi thứ nghe có vẻ hợp lý, cho đến khi tôi nhìn vào throughput thực tế.

Vấn đề gặp phải: 3 worker, nhưng chỉ 1 job inflight

Triệu chứng rất cụ thể. Supervisord báo cả 3 process đều "RUNNING". CloudWatch của SQS thì cho thấy ApproximateNumberOfMessagesVisible tồn đọng khá nhiều. Nhưng ApproximateNumberOfMessagesNotVisible (số message đang inflight) gần như lúc nào cũng dao động quanh 1. Nghĩa là tại một thời điểm, hệ thống của tôi thực chất chỉ xử lý được một job, dù tôi đã trả tiền cho ba worker.

Để hiểu tại sao, cần quay lại bản chất của FIFO, và tôi thấy một hình ảnh so sánh giúp mọi thứ sáng tỏ hơn bất kỳ định nghĩa nào.

Hãy tưởng tượng Message Group ID (MGID) như một làn thu phí trên cao tốc. FIFO cam kết rằng các xe trong cùng một làn sẽ đi qua đúng thứ tự, và quan trọng hơn: một chiếc xe phải qua trạm hoàn toàn thì chiếc kế tiếp mới được nhích lên. Trong ngôn ngữ SQS, "qua trạm hoàn toàn" nghĩa là message được delete sau khi xử lý xong, hoặc visibility timeout của nó hết hạn.

Vấn đề của tôi nằm ở chỗ: toàn bộ job "gửi thông báo giao dịch" đều được đẩy vào cùng một MGID. Tức là tôi đã dồn tất cả xe vào đúng một làn. Lúc này, việc thuê thêm hai nhân viên thu phí (hai worker) chẳng giải quyết được gì, vì làn đường chỉ cho một xe đi qua tại một thời điểm. Hai worker còn lại chỉ đứng nhìn.

Đây chính là đặc tính cốt lõi cần nhớ: khi một message trong một MGID đang inflight, không worker nào có thể poll được message mới thuộc cùng MGID đó cho đến khi message kia được xử lý xong. FIFO ưu tiên tính đúng thứ tự hơn tính song song, và đó là một sự đánh đổi có chủ đích của AWS.

Nhìn lại, throughput của tôi không bị giới hạn bởi số worker, mà bị giới hạn bởi số lượng MGID có message đang chờ. Với một MGID duy nhất, độ song song tối đa luôn là 1.

Phương án tối ưu throughput với nhiều MGID

Nếu một làn đường chỉ cho qua một xe, giải pháp hiển nhiên là mở nhiều làn. Trong SQS FIFO, "mở nhiều làn" chính là phân bổ message ra nhiều MGID khác nhau. Các message thuộc các MGID khác nhau được xử lý song song hoàn toàn độc lập, trong khi vẫn giữ nguyên thứ tự bên trong từng nhóm.

Ban đầu tôi định làm đơn giản: sinh ngẫu nhiên một MGID trong khoảng 100 giá trị cho mỗi lần dispatch. Cách này chạy được và ngay lập tức cho thấy cả 3 worker đều bận. Tuy nhiên, khi ngồi lại tôi nhận ra một điểm cần cân nhắc kỹ hơn về mặt thiết kế.

Chọn MGID theo khóa phân vùng tự nhiên, đừng chọn ngẫu nhiên

MGID không chỉ là công cụ để chia tải, nó còn là ranh giới đảm bảo thứ tự. Nếu tôi rải message hoàn toàn ngẫu nhiên, tôi vô tình từ bỏ luôn thứ tự giữa các message vốn cần đi cùng nhau.

Với bài toán thông báo giao dịch, câu hỏi quan trọng là: thứ tự cần được giữ trong phạm vi nào? Câu trả lời của tôi là "trong phạm vi một người dùng". Thông báo của user A và user B hoàn toàn độc lập, nhưng hai thông báo liên tiếp của cùng user A thì nên giữ đúng thứ tự.

Vậy nên thay vì random, tôi hash user_id (hoặc một khóa phân vùng tự nhiên tương tự) vào N bucket. Cách này cho tôi cả hai thứ: thứ tự được bảo toàn trong từng user, và độ song song trải rộng trên toàn bộ tập user.

use Illuminate\Support\Facades\Bus;

// Số lượng "làn" muốn mở. 100 là điểm khởi đầu hợp lý,
// có thể nâng lên khi số task tăng.
const MGID_BUCKETS = 100;

$groupId = 'transaction-notify:' . (crc32((string) $user->id) % MGID_BUCKETS);

TransactionNotificationJob::dispatch($payload)
    ->onQueue('notifications.fifo')
    ->withMessageGroupId($groupId);

Ghi chú: cách gắn Message Group ID có thể khác nhau tùy phiên bản package SQS FIFO bạn dùng cho Laravel. Điểm mấu chốt cần nắm là: MGID phải được suy ra từ khóa phân vùng, chứ không phải là một hằng số duy nhất cho mọi message.

Nếu tác vụ của bạn thực sự không cần thứ tự giữa các message (mỗi message hoàn toàn độc lập), thì rải ngẫu nhiên ra N MGID là chấp nhận được. Nhưng trong đa số trường hợp nghiệp vụ, luôn tồn tại một khóa tự nhiên đáng để dùng.

Độ song song thực tế bị chặn bởi cái gì?

Đây là điểm tôi muốn nhấn mạnh, vì lúc đầu tôi kỳ vọng sai. Sau khi có 100 MGID, tôi tưởng mình sẽ chạy được 100 job song song. Thực tế không phải vậy.

Để hiểu tại sao, tôi đi thẳng vào source code của Laravel. Mỗi vòng lặp của worker, driver SQS gọi phương thức pop(), và đây là toàn bộ phần lấy message (trích từ Illuminate\Queue\SqsQueue nhánh 12.x):

public function pop($queue = null)
{
    $response = $this->sqs->receiveMessage([
        'QueueUrl' => $queue = $this->getQueue($queue),
        'AttributeNames' => ['ApproximateReceiveCount'],
    ]);

    if (! is_null($response['Messages']) && count($response['Messages']) > 0) {
        return new SqsJob(
            $this->container, $this->sqs, $response['Messages'][0],
            $this->connectionName, $queue
        );
    }
}

Có hai chi tiết quyết định ở đây:

1. Lời gọi receiveMessage không truyền tham số MaxNumberOfMessages. Mà theo tài liệu của AWS SQS, giá trị mặc định của tham số này là 1 (khoảng hợp lệ từ 1 tới 10). Nghĩa là mỗi lần poll, SQS trả về tối đa đúng một message.

2. Ngay cả khi có nhiều message trả về, Laravel cũng chỉ lấy phần tử đầu tiên: $response['Messages'][0]. Nó bọc message đó thành một SqsJob, xử lý xong rồi mới quay lại vòng lặp để poll lần kế tiếp.

Nói cách khác, một tiến trình queue:work là một vòng lặp tuần tự: poll một message, xử lý, rồi mới poll message tiếp theo. Một worker không bao giờ ôm nhiều job cùng lúc. Đây chính là lý do vì sao số worker process trở thành một cận trên cứng cho độ song song, và nó dẫn thẳng tới công thức bên dưới.

Nguồn tham khảo:

• https://github.com/laravel/framework/blob/12.x/src/Illuminate/Queue/SqsQueue.php#L289-L308

• https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html#SQS-ReceiveMessage-request-MaxNumberOfMessages

Độ song song thực tế bằng:

concurrency = min(số worker process, số MGID đang có message chờ)

Với 3 process trên EC2, dù tôi có 100 hay 1000 MGID thì trần vẫn là 3 job đồng thời. 100 MGID không phải để chạy 100 job ngay lập tức, mà là để gỡ bỏ nút thắt nhân tạo, cho phép 3 worker được làm việc cùng lúc thay vì tranh nhau một làn duy nhất. Quan trọng hơn, nó tạo ra dư địa (headroom): khi tôi tăng số lượng consumer lên sau này, đã có sẵn nhiều làn để chúng cùng lấy việc.

Chính nhận thức này dẫn tôi tới bước tiếp theo. Muốn vượt qua trần "3 job", tôi phải scale số lượng consumer. Và scale worker trên một con EC2 với Supervisord thì đụng ngay trần CPU và bộ nhớ của chính con máy đó. Đã đến lúc nghĩ lại về hạ tầng.

Một lưu ý về "Exactly Once" mà tôi muốn thành thật

Trước khi đi tiếp, tôi cần đính chính chính giả định ban đầu của mình, vì tôi từng nghĩ "dùng FIFO thì khỏi cần idempotency". Điều đó chỉ đúng một phần.

Cơ chế khử trùng lặp của FIFO hoạt động trong một cửa sổ deduplication 5 phút (dựa trên MessageDeduplicationId hoặc content-based dedup). Nó ngăn việc cùng một message bị gửi lặp trong 5 phút đó. Nhưng nó không ngăn được việc một message bị giao lại (redeliver) nếu worker crash hoặc xử lý quá lâu khiến visibility timeout hết hạn trước khi kịp delete. Trong tình huống đó, message sẽ hiện ra trở lại và được xử lý một lần nữa.

Nói cách khác, FIFO cho tôi "exactly once" ở happy path và chống được việc submit trùng, nhưng nó không phải tấm khiên tuyệt đối trước redelivery khi có sự cố. Với tác vụ nhạy cảm như thanh toán, idempotency ở tầng ứng dụng vẫn là lớp bảo hiểm đáng có. Ở giai đoạn này tôi chấp nhận đánh đổi đó, nhưng tôi ghi nó vào danh sách nợ kỹ thuật một cách có ý thức, chứ không phải nhắm mắt cho qua.

Migrate sang Serverless với ECS Fargate

Con EC2 cùng Supervisord đã phục vụ tốt, nhưng nó có mấy điểm khiến tôi mệt: tôi phải tự vá lỗi OS, tự lo capacity, và mỗi lần muốn thêm worker là phải đụng vào cấu hình máy hoặc dựng thêm instance. Việc scale gắn chặt với vòng đời của một con máy vật lý (dù là ảo) làm mọi thứ kém linh hoạt.

Tôi quyết định đóng gói worker thành container và cho nó chạy trên ECS Fargate. Ý tưởng rất đơn giản: mỗi task Fargate là một consumer chạy php artisan queue:work. Muốn nhiều consumer hơn thì chỉ cần tăng desiredCount, không cần quan tâm tới máy chủ bên dưới nữa. AWS lo phần compute.

Lợi ích tôi nhận được:

Không còn quản lý máy chủ: không SSH, không vá OS, không lo Supervisord chết. Scale theo chiều ngang tự nhiên: một task là một đơn vị scale, tăng giảm gọn gàng. Cô lập và triển khai sạch: mỗi lần deploy là một image mới, rollback dễ. Trả tiền theo task đang chạy, thay vì nuôi một con EC2 luôn bật kể cả khi hàng đợi rỗng.

Điều tôi suýt làm sai: graceful shutdown

Đây là phần dễ bị bỏ qua nhất nhưng lại liên quan trực tiếp tới tính đúng đắn của dữ liệu. Khi ECS scale-in hoặc deploy phiên bản mới, nó gửi tín hiệu SIGTERM cho container, chờ một khoảng stopTimeout, rồi mới SIGKILL.

Nếu tôi để mặc định và job đang chạy dở bị SIGKILL, message chưa kịp delete sẽ quay lại hàng đợi và bị xử lý lại: đúng cái kịch bản redelivery mà tôi vừa nói ở trên. Với thông báo giao dịch, đó là gửi trùng.

May mắn là queue:work của Laravel xử lý SIGTERM rất ổn, khi nhận tín hiệu, nó hoàn tất job hiện tại rồi mới thoát (graceful). Việc của tôi là đảm bảo hai điều:

1. Tiến trình php artisan queue:work phải là tiến trình nhận tín hiệu trực tiếp trong container (chú ý vấn đề PID 1 và signal forwarding).

2. Đặt stopTimeout trong task definition đủ dài để job dài nhất kịp hoàn thành trước khi bị SIGKILL.

Một phần task definition rút gọn để minh họa:

{
  "family": "queue-worker-notifications",
  "requiresCompatibilities": ["FARGATE"],
  "cpu": "1024",
  "memory": "2048",
  "taskRoleArn": "arn:aws:iam::<account-id>:role/queue-worker-sqs-access",
  "containerDefinitions": [
    {
      "name": "queue-worker",
      "image": "<account-id>.dkr.ecr.<region>.amazonaws.com/app:latest",
      "essential": true,
      "command": ["php", "artisan", "queue:work", "sqs", "--tries=3", "--max-time=3600"],
      "stopTimeout": 120,
      "logConfiguration": {
        "logDriver": "awslogs",
        "options": {
          "awslogs-group": "/ecs/queue-worker-notifications",
          "awslogs-region": "<region>",
          "awslogs-stream-prefix": "worker"
        }
      }
    }
  ]
}

Vài điểm đáng chú ý: quyền truy cập SQS được cấp qua task role (IAM) thay vì hardcode credential; log được đẩy thẳng lên CloudWatch Logs; và --max-time giúp worker tự khởi động lại định kỳ, tránh rò rỉ bộ nhớ vốn là bệnh kinh niên của các long-running PHP process.

Scale-out/in theo Backlog trên mỗi Task

Đóng gói thành Fargate mới chỉ giải quyết được "làm sao thêm consumer dễ dàng". Câu hỏi tiếp theo khó hơn: thêm bao nhiêu, và khi nào? Đặt cứng desiredCount = 10 thì lãng phí lúc rỗng và vẫn thiếu lúc cao điểm. Tôi cần một tín hiệu phản ánh đúng áp lực thực tế của hệ thống.

CPU và memory không phải tín hiệu tốt ở đây, vì worker gửi thông báo phần lớn thời gian là chờ I/O, chứ không ngốn CPU. Tín hiệu đúng đắn nhất chính là độ sâu hàng đợi trên mỗi task, hay còn gọi là Backlog per Task.

Ý tưởng nền tảng rất trực quan: điều tôi thực sự quan tâm không phải "hàng đợi còn bao nhiêu message", mà là "với năng lực hiện tại, mỗi message phải chờ bao lâu mới tới lượt". Cứ tưởng tượng một siêu thị: cái quyết định bạn có mở thêm quầy tính tiền hay không, không phải tổng số khách trong siêu thị, mà là số khách đang xếp hàng chia cho số quầy đang mở.

Công thức

Đầu tiên, tôi xác định một ngưỡng backlog chấp nhận được cho mỗi task, dựa trên độ trễ mà nghiệp vụ chịu được và thời gian xử lý trung bình một message:

acceptableBacklogPerTask = acceptableLatencySeconds / avgProcessingSecondsPerMessage

Ví dụ tôi chấp nhận thông báo trễ tối đa 30 giây, và mỗi message xử lý mất khoảng 0.5 giây, thì:

acceptableBacklogPerTask = 30 / 0.5 = 60 message/task

Sau đó, metric mà tôi theo dõi theo thời gian thực là:

backlogPerTask = ApproximateNumberOfMessagesVisible / RunningTaskCount

Chính sách Target Tracking Scaling của Application Auto Scaling sẽ cố gắng giữ backlogPerTask bám sát mục tiêu 60. Khi hàng đợi dồn lên, tỷ số này tăng vượt 60, ECS tự thêm task. Khi hàng đợi vơi, tỷ số giảm dưới 60, ECS tự bớt task.

Cách tính metric

ApproximateNumberOfMessagesVisible có sẵn từ SQS và RunningTaskCount có sẵn từ ECS, nhưng bản thân tỷ số giữa hai metric thì không tồn tại sẵn. Có hai cách phổ biến để tạo ra nó:

1. CloudWatch Metric Math: dùng biểu thức toán học kết hợp hai metric để tạo metric phái sinh, rồi target track trên metric đó. Cách này gọn, không cần thêm hạ tầng.

2. Lambda định kỳ publish custom metric: một Lambda chạy mỗi phút, đọc hai giá trị, tính tỷ số và đẩy lên CloudWatch dưới dạng custom metric. Cách này linh hoạt và dễ kiểm soát, phù hợp khi logic tính toán phức tạp hơn.

Tôi ưu tiên CloudWatch Metric Math, vì nó gọn và không phải nuôi thêm một Lambda nào cả. Target Tracking của Application Auto Scaling cho phép khai báo một customized metric specification dựa trên metric math, trong đó ta lấy hai metric gốc rồi tính tỷ số ngay trong biểu thức. Một điểm khéo là dùng MAX([tasks, 1]) để tránh chia cho 0 khi không có task nào đang chạy.

Cấu hình target tracking (rút gọn) trông như sau:

{
  "TargetValue": 60.0,
  "CustomizedMetricSpecification": {
    "Metrics": [
      {
        "Id": "visible",
        "MetricStat": {
          "Metric": {
            "Namespace": "AWS/SQS",
            "MetricName": "ApproximateNumberOfMessagesVisible",
            "Dimensions": [{ "Name": "QueueName", "Value": "notifications.fifo" }]
          },
          "Stat": "Average"
        },
        "ReturnData": false
      },
      {
        "Id": "tasks",
        "MetricStat": {
          "Metric": {
            "Namespace": "ECS/ContainerInsights",
            "MetricName": "RunningTaskCount",
            "Dimensions": [
              { "Name": "ClusterName", "Value": "queue-workers" },
              { "Name": "ServiceName", "Value": "notifications-worker" }
            ]
          },
          "Stat": "Average"
        },
        "ReturnData": false
      },
      {
        "Id": "backlogPerTask",
        "Expression": "visible / MAX([tasks, 1])",
        "ReturnData": true
      }
    ]
  }
}

Lưu ý: RunningTaskCount xuất hiện dưới namespace ECS/ContainerInsights, nên bạn cần bật Container Insights cho cluster để metric math có thể tham chiếu tới nó.

Cách Lambda vẫn hữu ích khi logic tính toán phức tạp hơn (ví dụ muốn kết hợp thêm số message inflight, hay làm mượt giá trị theo thời gian), nên tôi để lại đây một bản mẫu để bạn tham khảo:

import boto3

sqs = boto3.client("sqs")
ecs = boto3.client("ecs")
cw = boto3.client("cloudwatch")

def handler(event, context):
    attrs = sqs.get_queue_attributes(
        QueueUrl=QUEUE_URL,
        AttributeNames=["ApproximateNumberOfMessages"],
    )
    backlog = int(attrs["Attributes"]["ApproximateNumberOfMessages"])

    svc = ecs.describe_services(cluster=CLUSTER, services=[SERVICE])
    running = svc["services"][0]["runningCount"]

    # Tránh chia cho 0: coi như đang có tối thiểu 1 task.
    per_task = backlog / max(running, 1)

    cw.put_metric_data(
        Namespace="Custom/QueueWorker",
        MetricData=[{
            "MetricName": "BacklogPerTask",
            "Value": per_task,
            "Unit": "Count",
        }],
    )

Những cái bẫy khi scale

Có ba điều tôi học được sau khi cho pipeline này chạy thật:

Chống flapping (rung lắc): nếu scale-in quá nhạy, hệ thống sẽ liên tục tăng rồi giảm task một cách vô nghĩa. Đặt cooldown cho scale-in dài hơn scale-out để hệ thống "nhả phanh từ từ", ưu tiên phản ứng nhanh khi tải lên và thận trọng khi tải xuống.

Bài toán scale từ 0: nếu để minCapacity = 0 cho tiết kiệm, bạn sẽ gặp vấn đề. Khi RunningTaskCount = 0, metric backlogPerTask không xác định (chia cho 0), nên Target Tracking không có cơ sở để bật task đầu tiên. Giải pháp là hoặc giữ minCapacity = 1, hoặc dùng một alarm riêng trên ApproximateNumberOfMessagesVisible > 0 để kích task đầu tiên từ 0 lên 1, rồi để Target Tracking lo phần còn lại.

Đừng quên trần MGID: nhớ lại phần 2, độ song song bị chặn bởi số MGID có message chờ. Nếu bạn chỉ có 100 MGID mà lại scale lên 200 task, thì 100 task sẽ đói việc. Số MGID cần được thiết kế đủ lớn để phủ được số task tối đa mà bạn dự định scale tới. Đây là lý do tôi chọn con số 100 (hoặc hơn) ngay từ đầu: nó là dư địa cho việc scale sau này.

Kết luận và hướng cải thiện tiếp theo

Nhìn lại cả hành trình, bài học lớn nhất với tôi không phải là một dòng lệnh hay một cấu hình cụ thể, mà là cách đọc đúng nút thắt của hệ thống.

Nút thắt ban đầu không nằm ở số worker, mà ở việc mọi message dồn vào một MGID duy nhất. FIFO đánh đổi tính song song để lấy thứ tự, và tôi cần tôn trọng sự đánh đổi đó thay vì chống lại nó. Nhiều MGID giúp mở làn, nhưng MGID nên bắt nguồn từ khóa phân vùng tự nhiên để vừa song song vừa giữ thứ tự trong từng thực thể. Fargate gỡ bỏ gánh nặng quản lý máy chủ và biến "thêm consumer" thành một thao tác đơn giản, với điều kiện phải làm đúng graceful shutdown. Cuối cùng, Backlog per Task là tín hiệu scale phản ánh đúng trải nghiệm người dùng hơn hẳn CPU hay memory.

Và cũng cần nói thẳng: FIFO không cho tôi "exactly once" tuyệt đối. Nó rất tốt trong điều kiện bình thường, nhưng idempotency ở tầng ứng dụng vẫn là lớp phòng thủ mà một hệ thống thanh toán nghiêm túc nên có.

Về những bước tiếp theo, đây là danh sách tôi đang cân nhắc:

• Bổ sung idempotency thật sự cho các tác vụ nhạy cảm, dùng một khóa tự nhiên (ví dụ transaction ID) làm khóa chống trùng, để an tâm trước mọi kịch bản redelivery.

• Cấu hình Dead Letter Queue (DLQ) cho FIFO, để những message thất bại nhiều lần không kẹt lại làm nghẽn cả một MGID, kéo theo toàn bộ nhóm bị đình trệ.

• Đánh giá chế độ High Throughput cho FIFO, khi lượng message tiếp tục tăng và tôi chạm tới trần throughput của FIFO tiêu chuẩn.

• Tinh chỉnh số bucket MGID và ngưỡng backlog dựa trên dữ liệu vận hành thật, thay vì các con số phỏng đoán ban đầu.

Nếu bạn đang ở giai đoạn "3 worker mà chỉ 1 job chạy", hy vọng câu chuyện này giúp bạn tìm ra nút thắt của riêng mình nhanh hơn. Nếu bạn có cách tiếp cận khác cho bài toán throughput trên FIFO, tôi rất mong được nghe và học hỏi thêm.

Cảm ơn các bạn đã đọc tới đây.