Ánh xạ nguồn sự kiện và xử lý luồng

Ứng dụng Serverless với AWS Lambda

Claudio Canales

Senior DevOps Engineer

Ánh xạ nguồn sự kiện nằm ở đâu

Hàng đợi và luồng thường được thăm dò theo lô.
Ánh xạ nguồn sự kiện quản lý việc thăm dò và gom lô.

Vị trí ánh xạ nguồn sự kiện

Ánh xạ nguồn sự kiện là gì?

Tài nguyên kết nối một nguồn với hàm của bạn.
Lambda thăm dò nguồn để lấy record.
Lambda gọi handler của bạn với một lô.

Luồng ánh xạ nguồn sự kiện

Đẩy vs thăm dò (tại sao quan trọng)

Mô hình đẩy (Push)

Nguồn đẩy gửi sự kiện ngay lập tức.
Ví dụ: tải lên Amazon S3.

Mô hình thăm dò (Poll)

Nguồn thăm dò được Lambda kiểm tra.
Lambda gom lô và gọi handler của bạn.
Điều này ảnh hưởng độ trễ và cách retry.

Mô hình đẩy vs thăm dò

Hàng đợi vs luồng

Hàng đợi (Amazon SQS)

Hàng đợi thông điệp được quản lý, như hộp thư đến.

Luồng (DynamoDB Streams)

Nhật ký thay đổi của một bảng DynamoDB.
Cả hai đến dưới dạng Records, nhưng ý nghĩa khác nhau.

So sánh hàng đợi và luồng

Dạng sự kiện theo lô (đơn giản hóa)

{
  "Records": [{
    "messageId": "abc-123",
    "body": "{\"order_id\": \"A-42\"}"
  }]
}

Hầu hết sự kiện ánh xạ bắt đầu với Records.
Mỗi record có messageId và chuỗi body.
Phân tích body thành payload của bạn.

Hướng dẫn: xử lý từng record

def lambda_handler(event, context):
    records = event.get("Records", [])
    for record in records:
        body = record.get("body", "")
        print("BODY:", body)
    return {"statusCode": 200}

Đọc Records với danh sách mặc định.
Lặp qua từng record và đọc body an toàn.
Ghi log cần thiết, rồi trả về.

Hướng dẫn: phân tích JSON trong body

import json

def lambda_handler(event, context):
    record = event.get("Records", [])[0]
    payload = json.loads(record.get("body", "{}"))
    order_id = payload.get("order_id")
    print("ORDER_ID:", order_id)
    return {"statusCode": 200}

Import json và đọc body với mặc định an toàn.
Phân tích bằng json.loads để lấy dict.
Trích xuất trường cần dùng; luôn kiểm tra trước.

Gom lô: kích thước và cửa sổ lô

Kích thước lô: số record mỗi lần gọi.
Cửa sổ lô: thời gian Lambda chờ để đủ lô.
Lô lớn tăng thông lượng nhưng thêm việc mỗi lần chạy.
Với SQS, theo dõi ApproximateAgeOfOldestMessage để biết tồn đọng.

Biểu đồ đánh đổi kích thước lô

Đánh đổi kích thước lô

Lô lớn hiệu quả nhưng có thể tăng độ trễ.
Lô nhỏ giảm thời gian xử lý đầu tiên nhưng tăng số lần gọi.
Chọn theo khối lượng công việc.

Kích thước lô vs độ trễ

Mở rộng bằng đồng thời

Nhiều lô được xử lý đồng thời.
Độ đồng thời cao hơn tăng thông lượng.
Nhưng cũng tăng tải xuống hạ nguồn.

Xử lý lô đồng thời

Bảo vệ hệ thống hạ nguồn

Quá nhiều đồng thời có thể làm quá tải cơ sở dữ liệu hoặc API.
Dùng giới hạn để kiểm soát tải.
Ưu tiên đơn vị công việc nhỏ, an toàn.

Bảo vệ tải hạ nguồn

Lỗi lô một phần

Một lô có thể chứa record tốt và xấu.
Không có lỗi một phần, một record xấu sẽ retry cả lô.
Xử lý lỗi một phần cho phép chỉ retry mục lỗi.

Luồng lỗi lô một phần

Hướng dẫn: báo lỗi một phần (SQS)

def lambda_handler(event, context):
    failures = []
    for record in event.get("Records", []):
        try:
            process(record)
        except Exception:
            failures.append({"itemIdentifier": record["messageId"]})
    return {"batchItemFailures": failures}

Thu thập messageId của các record lỗi trong batchItemFailures.
Lambda chỉ retry các mục đó.
Record xử lý thành công sẽ không gửi lại.

Điểm chính

Ánh xạ nguồn sự kiện thăm dò hàng đợi và luồng.
Chúng chuyển Records theo lô cho handler.
Kích thước và cửa sổ lô đánh đổi độ trễ lấy thông lượng.
Lỗi một phần tránh xử lý lại mục đã thành công.

Sơ đồ tóm tắt ánh xạ

Ayo berlatih!

Ứng dụng Serverless với AWS Lambda