Tổng quan về SQS
Amazon Simple Queue Service (SQS) là gì ?
SQS là dịch vụ hàng đợi thông điệp phân tán, được giới thiệu bởi Amazon Web Service năm 2004. Là dịch vụ hỗ trợ gửi và nhận thông điệp thông qua môi trường internet HTTP API. SQS được thiết kế để nhắm đến các ứng dụng cần khả năng mở rộng cao. Bởi vì cung cấp HTTP API nên các thành phần khác nhau của hệ thống có thể gửi nhận thông điệp một cách dễ dàng.
Một vài đặc điểm nổi bật của SQS
Tính mở rộng cao
SQS mở rộng bởi dung lượng đọc/ghi của thông điệp, nên chúng ta không cần quan tâm đến việc mở rộng hàng đợi, bởi việc này đã có AWS lo.
Trả tiền trên số lần xử lý
Chúng ta phải trả tiền cho mỗi lượt đọc/ghi thông điệp, không có khoản phí định kỳ nào.
Dễ dàng thiết lập
Khi lựa chọn SQS, không cần phải setup bất kỳ hạ tầng nào để có thể sử dụng nó. Chỉ đơn giản là gọi API để đọc/ghi thông điệp.
Có hai lựa chọn là Standard và FIFO queues
Có hai lựa chọn để thiết lập queue là tiêu chuẩn hay FIFO, tùy theo nhu cầu sử dụng mà lựa chọn cho phù hợp.
Tự động xử lý thông điệp trùng lặp ở FIFO queues
Xử lý lặp thông điệp khá quan trọng trong queue, SQS sẽ xóa những thông điệp trùng lặp. Điều này làm FIFO queue phù hợp cho các tác vụ mà mỗi task chỉ chạy một lần.
Xử lý lỗi qua Dead letter queue
Chức năng quan trọng để debug lỗi. Tất cả các thông điệp mà không thể xử lý sẽ được lưu vào Dead letter queue, nó như là một hàng đợi bình thường và chúng ta có thế kiểm tra hoặc xử lý logic cho những thông điệp lỗi trên Dead letter queue này.
SQS hoạt động như thế nào ?
Behind the scenes:
Hàng đợi (queue) hoạt động như thế nào ?
Điển hình thì xử lý hàng đợi sẽ như sau. Hệ thống có các Producer với vai trò là đấy thông điệp vào hàng đợi, các Consumer sẽ nhận thông điệp và xử lý chúng.
Hệ thống đánh dấu là tin nhắn đang pending
hoặc là in-flight
và gửi nó đến các consumer
- các background job management ở consumer client sẽ chạy và xử lý các thông điệp này.
Khi mà consumer
hoàn thành xử lý, nó sẽ thông báo cho queue
là đã xử lý thành công bằng cách gửi một thông báo acknowledge
- xác nhận xử lý thành công. Sau đó thì thông điệp được đánh dấu từ pending
thành processed
.
Nếu quá trình này xảy ra break ở đâu đó, ví dụ thiếu thông báo acknowledge
trong một khoảng thời gian nhất định chẳng hạn, thì có nghĩa là quá trình xử lý thông điệp không thành công, và thông điệp không còn được coi là pending
nữa. Như vậy sẽ không có thông điệp nào bị kẹt ở trạng thái pending
mãi mãi.
Khi mà xử lý fail một số lần nhất định, queue sẽ coi là các thông điệp này có vấn đề và ngừng gửi nó đến các consumer
. Hoặc cũng có thể đưa nó lưu vào Dead letter queue (DLQ).
Với SQS
SQS cung cấp một api endpoint để các producer
gửi thông điệp vào queue
, các consumer
sẽ đọc thông điệp và xử lý thông điệp đó.
- Thông điệp gửi đến SQS có thể là chuỗi string bất kỳ, XML hay là JSON.
- SQS không giới hạn số lượng thông điệp được gửi đến và được xử lý.
Thông điệp
Một thông điệp ở SQS sẽ trông như này.
{
"message_id": "b1438d18-8095-4bed-ad26-ee1b8768e537",
"receipt_handle": "AQEBLUaoBHhmJXetoonDkqfDDHUSlkGsIXLtHNATnSCeDd2nzk6V45nTnOt2PSCjqt5bbm2uCBUdoablpUJ59q4qS/kWOoT6h/R6y+kax+H6GNasKX6NU7IzGg/kr2E+QiI/OssES0niKE7J45fAge7UnrafrBWbpo1vJCgJzEzot3Z00AvwhsYmqxIiItQV7TRz165gdf9MTD+uILt6WNboJqglKpjba/4SN1MbQtuI7fEO0TFbp673YmNG8haNi0epfofzMKa+90+pjCDHBjUmSxkCrbx42Q3JIgA4LWPQVoBUo6mwttLyfkcSzoDfHVbGl0/Aph6uMgMymDkfN++79Il/LCrwOHSHj9Mpv1C1AuXj+TGO+6Vu1qvFvWdmoIvHNR+/AvALxWLKkhUH+ky9yg==",
"md5_of_body": "c9f0f895fb98ab9159f51fd0297e236d",
"body": "8",
"attributes": {
"SenderId": "596865106215",
"ApproximateFirstReceiveTimestamp": "1605893094056",
"ApproximateReceiveCount": "1",
"SentTimestamp": "1605892972066"
},
"md5_of_message_attributes": null,
"message_attributes": {}
}
Mỗi thông điệp sẽ được gán một message_id
, receipt_handle
cũng như là body
, message_attributes
.
receipt_handle
sẽ dùng cho trường hợp muốn xóa thông điệp ở queue
Visible timeout
-
Visible timeout là thời gian chờ tối đa mà thông điệp được xử lý ở
consumer
-
Với một queue có visible timeout 60 phút thì khi một message được
consumer
đọc, và xử lý thì timer bắt đầu chạy, thông điệp đó sẽ coi như bị ẩn đi 60 phút - ở trạng tháiin-flight
, các consumer khác không đọc được thông điệp đó trong một khoảng thời gian. -
Nếu đang xử lý mà consumer client bị ngỏm, mất mạng hay vì lý do gì đó mà không thể xử lý, thì sẽ không thể tiếp tục thông báo tới queue về tình trạng xử lý. Vậy nên quá thời gian này thì thông điệp đó sẽ tiếp tục xuất hiện để consumers kéo về xử lý tiếp.
SQS tích hợp Ruby SDK
Quen thuộc với các dịch vụ khác của AWS, với SQS thì AWS cũng cung cấp đầy đủ api trong sdk để sử dụng. Sau đây sẽ là một vài ví dụ đơn giản:
Để sử dụng SQS sdk ruby thì chúng ta cần cài gem aws-sdk-sqs
, và cấu hình đầy đủ AWS với key id và acess key id.
Khởi tạo SQS với queue_name
SQS = Aws::SQS::Client.new(region: "us-east-1")
queue = SQS.create_queue(queue_name: "my_queue")
Tạo một message mới
SQS.send_message({queue_url: queue_url, message_body: "process_record_id", message_attributes: {"Id" => {string_value: "1", data_type: "String"}}})
Với poller thì chúng ta có thể lắng nghe queue liên tục để bắt message ở queue
QUEUE_URL = SQS.get_queue_url(queue_name: "my_queue").queue_url
POLLER = Aws::SQS::QueuePoller.new(QUEUE_URL)
poller_stats = POLLER.poll({max_number_of_messages: 10 }) do |messages|
messages.each do |message|
puts "Message body: #{message.body}"
# Do any logic process message here
end
end
Hoặc chủ động lấy message:
SQS.receive_message({
queue_url: QUEUE_URL,
max_number_of_messages: 5,
visibility_timeout: 60
})
Với mỗi message thì sau khi handle xong, chúng ta cần xóa nó ở queue để hoàn thành job, params cần có receipt_handle là value lúc chúng ta get message về
resp = SQS.delete_message({
queue_url: QUEUE_URL,
receipt_handle: "receipt_handle value",
})
Trường hợp trong lúc xử lý message, máy bị mất kết nối hoặc bị ngỏm, sau một số lần nhất định thì message sẽ được lưu vào Dead letter queue, việc chúng ta cần làm là tiếp tục handle những message lỗi này nếu cần.
Chúng ta cần phải tự tạo một queue để sử dụng làm Dead letter queue này, qua SDK hoặc qua AWS Console UI
- Sử dụng SDK:
SQS.create_queue({
queue_name: "my_dead_letter_queue"
})
dead_letter_queue_url = SQS.get_queue_url(queue_name: "my_dead_letter_queue").queue_url
dead_letter_queue_arn = SQS.get_queue_attributes({
queue_url: dead_letter_queue_url,
attribute_names: ["QueueArn"]
})
.attributes["QueueArn"]
redrive_policy = {
"maxReceiveCount" => "5", # After the queue receives the same message 5 times, send that message to the dead letter queue.
"deadLetterTargetArn" => dead_letter_queue_arn
}.to_json
# Sau đó set queue attributes
SQS.set_queue_attributes({
queue_url: QUEUE_URL,
attributes: {
"RedrivePolicy" => redrive_policy
}
})
- Với AWS console UI, chúng ta tạo một queue mới là dead_letter_queue chẳng hạn, sau đó vào edit queue chính, chọn enable dead letter và select queue dead_letter_queue là xong (easy).
- Để xóa queue:
SQS.delete_queue(queue_url: QUEUE_URL)
SQS tích hợp gem shoryuken
Với gem shoryuken, việc tích hợp background job sử dụng SQS đơn giản và dễ dàng hơn bao giờ hết.
https://github.com/phstc/shoryuken
Chúng ta cần cài 2 gem này vào nhé.
gem "shoryuken"
gem "aws-sdk-sqs"
Với những ai đã quen thuộc với bộ đôi Sidekiq + Redis thì combo shoryuken + SQS cũng khá là tương đồng, cơ chế cũng như cách sử dụng gần như là không khác gì nhau cả.
Lan man một tý:
Về combo Sidekiq + Redis thì chúng ta cần mua gói PRO (950 $ / năm) để có thể handle job failed như Dead letter bên SQS.
Còn với SQS thì job chạy sẽ không được realtime cho lắm vì thông qua HTTP API, sẽ không nhanh bằng combo Sidekiq + Redis, phù hợp cho những Job không cần realtime
Những Job như bắn thông báo, gửi notification hay cần ưu tiên xử lý nhanh thì nên dùng Redis vì nó bắt rất nhanh - ngược lại thì có thể sử dụng SQS.
- Khai báo worker
class HelloWorker
include Shoryuken::Worker
shoryuken_options queue: "my_queue", auto_delete: true
def perform(sqs_msg, data)
puts "Hello, #{data}"
end
end
- Start shoryuken
bundle exec shoryuken -q my_queue -R
- Và enqueue một job
HelloWorker.perform_async({message: "1"})
Như vậy là chúng ta vừa tìm hiểu về Amazon Simple Queue Service (SQS) qua một vài khái niệm, các đặc điểm nổi bật, cách sử dụng SQS tích hợp với SDK ruby cũng như sài gem. Hy vọng bài viết sẽ có ích cho những ai đang tìm hiểu về SQS
Cảm ơn mọi người đã theo dõi bài viết!