AWS SQS, SNS, ARN แต่ละตัวคืออะไร รู้จักผ่าน Event-Driven Archiecture (EDA) ด้วย Python + boto3

AWS system diagram (SNS, SQS) — Publisher ส่งครั้งเดียว + SNS กระจายไปทุก Queue โดยอัตโนมัติ, Consumer แต่ละตัวทำงานอิสระ

ขั้นตอนการทำงานของ small workshop ที่เราเล่นกันนะคับ

สรุปแบบง่ายที่สุดSNS = Pub/Sub, SQS = Queue , ARN = ID ของ resources in AWS (ไว้ใช้คุยกันภายใน)จบจ้า 55

• Simple project ที่เราจะทำกันนะคับ

publish_order.py    SNS Topic             SQS Queue                Consumer
     │                  │                     │
     │──── publish ────►│                     │
     │                  │── fan out ──► TestInventoryQueue ◄── process_inventory.py
     │                  │── fan out ──► TestEmailQueue     ◄── (❌ not implement yet)

🟠 สรุป SQS:
- 1 request = 64KB, ถ้าข้อความที่ส่งมามีขนาด 256KB (จะคิดเป็น 4 requests จ้า)
- 256KB = Maximum payload ที่ส่งได้นะ
- ข้อความจะค้างในคิวไปเรื่อยๆนะ จนกว่าจะมีโปรแกรมมาดึงและลบออก หรือ จนหมดอายุ 4 วัน (ตั้งได้)
- SQS ไม่มี log นะคับ "ลบแล้วลบเลย" เมื่อมีคนดึงไปทำงานและสั่ง delete แบบกู้คืนไม่ได้ เพราะ
  ต้องทำให้คิวว่าง และประมวลผลงานต่อไปให้เร็วที่สุด
  ถ้าต้องการเก็บ log ให้เก็บจากฝั่ง coding นะคับ (จะลงไฟล์หรือส่งไป ELK Stack ก็ได้)

สรุป
1. SNS Publish (1 ครั้ง) -> ฟรี
2. SNS Deliver ไปยัง 2 คิว -> ฟรี
3. SQS Receive & Delete สำหรับ 2 คิว -> รวมเป็น 4 Requests -> ฟรี

┌─────────────────────┬─────────────────────┬──────────────────────────────┐
│ Service             │ Free Tier/month     │ After free tier (ต่อ 1 ล้านครั้ง)│
├─────────────────────┼─────────────────────┼──────────────────────────────┤
│ SQS (Standard)      │ 1,000,000 Requests  │ ~$0.40 (ประมาณ 14 บาท)       │
├─────────────────────┼─────────────────────┼──────────────────────────────┤
│ SNS (Publish)       │ 1,000,000 Requests  │ ~$0.50 (ประมาณ 17.50 บาท)    │
├─────────────────────┼─────────────────────┼──────────────────────────────┤
│ SNS To SQS Delivery │ ไม่จำกัด (Unlimited)  │ FREE                         │
└─────────────────────┴─────────────────────┴──────────────────────────────┘

1. SNS (Simple Notification Service) = สำนักพิมพ์ (Publisher)

SNS ทำหน้าที่เป็นคนกระจายข่าวสารแบบ Publish/Subscribe (Pub/Sub) = สมมติว่า SNS คือสำนักพิมพ์ที่มีข่าวสารใหม่ๆ ออกมาตลอดเวลา

• ตัวมันเองจะไม่สนใจว่าใครจะอ่านบ้าง มันมีหน้าที่แค่ “ประกาศออกไป” ตามช่องทางต่างๆ ที่มีคนมาลงทะเบียนไว้
• ในระบบเราจะเรียกว่าช่องกระจายข่าวนี้ว่า Topic (เช่น หัวข้อข่าวบันเทิง, หัวข้อข่าวกีฬา)

2. SQS (Simple Queue Service) = ตู้จดหมายหน้าบ้าน (Queue)

SQS ทำหน้าที่เป็น คิวเก็บฝากข้อความ (Message Queue) แบบ Point-to-Point

• สมมติว่า SQS คือ “ตู้รับจดหมาย” ที่ตั้งอยู่หน้าบ้านของระบบต่างๆ (เช่น บ้านของระบบส่งอีเมล, บ้านของระบบตัดเงิน)
• เวลาส่งอะไรมา มันจะถูกเก็บไว้ในตู้จดหมายนี้อย่างปลอดภัย ไม่หายไปไหน แม้ว่าเจ้าของบ้าน (Server) จะง่วงนอน ปิดปรับปรุง หรือประมวลผลไม่ทันก็ตาม พอเจ้าของบ้านพร้อม ก็จะมาเปิดตู้ดึงจดหมายไปอ่านทีละฉบับ

3. ARN (Amazon Resource Name) = id number ของ all resources in AWS (แบบมี pattern)

เพราะ AWS มี services ต่างๆเยอะมากไง เช่น Queue in SQS, Topic in IoT Core, สิทธิ์ in IAM ฉะนั้นเลยต้องมีรูปแบบชื่อที่เป็นสากล เพื่อให้รู้ว่าเรากำลังสั่งการ หรือ คุยกับ service ไหนกันแน่

• เราจะ copy ไปใช้เมื่อต้องการเชื่อมโยง services ต่างๆเข้าด้วยกันเช่น

1. การตั้งค่าสิทธิ์ (IAM Policy): บอกระบบว่า “อนุญาตให้แอปพลิเคชันนี้ ส่งข้อมูลเข้า SQS Queue ตัวนี้เท่านั้น” เราก็เอา ARN ของ SQS คิวนั้นไปใส่ในช่อง Resource ใน Policy
2. การผูกบริการเข้าด้วยกัน: เช่น ถ้าอยากให้ AWS IoT Core ส่งข้อมูลต่อ突เข้า SQS ทันทีที่ได้รับข้อความ ในหน้าตั้งค่า Rule ของ IoT Core เราก็ต้องระบุ ARN ของ SQS ปลายทางลงไป เพื่อให้มันส่งของไปถูกบ้านครับ

พูดง่ายๆ คือ ถ้าเมื่อไหร่ที่บริการของ AWS ต้องคุยกัน หรือต้องการอ้างอิงถึงกัน ARN คือสิ่งที่คุณต้องหยิบมาใช้เสมอครับ

# arn:partition:service:region:account-id:resource-type/resource-id

arn:aws:sqs:us-east-1:123456789012:MyShoppingQueue

=================================================
# Explain
arn: จุดเริ่มต้น เพื่อบอกว่านี่คือ ARN นะ
aws: Partition (ส่วนใหญ่คือ aws แต่ถ้าเป็นโซนจีนจะเป็น aws-cn)
sqs: ชื่อบริการ (Service) ว่าเป็นของ SQS
us-east-1: Region หรือภูมิภาคที่คิวนี้ตั้งอยู่
123456789012: AWS Account ID (เลขบัญชี AWS ของเรา 12 หลัก)
MyShoppingQueue: ชื่อของตัว Queue นั้นๆ (Resource ID)

ดูรูปนี้ก็ได้ ที่เราเล่นกันจริงๆ จะเห็นว่า SNS, SQS connect กันด้วน ARM นะ

นี่แค่เร่ิมต้นจ้า 3 สาวมาให้กำลังใจ

เรามาลองสร้าง Simple projct กันเล่นๆดีกว่านะคือ “ระบบประมวลผลคำสั่งซื้อ E-commerce (Order Processing System)”

พอมีลูกค้ากดสั่งซื้อของในเว็บ จะมี 2 tasks ตามมาคือ

1. ตัด stocks สินค้า (Inventory)
2. ส่ง email ยืนยันการสั่งซื้อ (Email Notification)

แน่นอนเราต้องการให้ระบบ 2 อันนี้ทำงานเป็น Async นะ ไม่งั้นถ้าอะไรพัง อีกอันจะพังไปด้วย

Let’s get started

1. Create SQS (สร้างคิว มารับงานไปทำจ้า)

# Standard vs FIFO

Standard = เร็วสุด 🚀 (High Throughput)
           แต่ "อาจซ้ำ" และ "ลำดับไม่เป๊ะ" (coding ต้อง handle duplicate case เอง)
           เหมาะกับ ส่ง email, event processing ทั่วไป, log processing

FIFO  = ถูกต้องสุด ✅
        ไม่ซ้ำ และ ลำดับเป๊ะ (1-2-3-4)
        แต่ช้ากว่า ได้ที่ (300-3000 req/sec)
        เหมาะกับ ระบบการเงิน, ระบบตัด stock สินค้า, order processing, workflow ที่สำคัญ

# Infor from 2 SQS and 1 SNS

# TestInventoryQueue (SQS 1)
URL:
https://sqs.ap-southeast-7.amazonaws.com/154230581564/TestInventoryQueue

ARN:
arn:aws:sqs:ap-southeast-7:154230581564:TestInventoryQueue

# TestEmailQueue (SQS 2)
URL:
https://sqs.ap-southeast-7.amazonaws.com/154230581564/TestEmailQueue

ARN:
arn:aws:sqs:ap-southeast-7:154230581564:TestEmailQueue

# TestOrderCreatedTopic (SNS - มีแค่ ARN นะ, ไมมี URL คุยกันภายในเท่านั้น)
# เป็นตัวรับข้อความทีส่งเข้ามานะ ก่อนจะ fan out ไปให้ SQS แต่ละตัว ตามที่เราได้ตั้งเอาไว้
ARN:
arn:aws:sns:ap-southeast-7:154230581564:TestOrderCreatedTopic

#############################################################
ARN ใช้สำหรับ IAM policies, SNS subscription (คุยภายใน AWS)
URL ใช้สำหรับ API calls จริงๆ เช่น ReceiveMessage, DeleteMessage
    (ใครมี credentials ก็ยิง HTTP มาได้เลย ไม่ต้องอยู่ใน AWS)

Create SQS name “TestInventoryQueue”

ได้มาแล้ว TestInventoryQueue

• จากนั้นเราจะสร้างอีก queue นึงใช้ชื่อว่า **TestEmailQueue** นะ

เรียบร้อยล่ะ ได้ Queue ครบแล้ว 2 ตัว (message in queue ควรเป็น 0 ตลอดนะ หลัง process เสร็จ, ไม่ควรค้างจ้า)

2. สร้าง SNS Topic (สร้างโทรโข่ง)

Create SNS name “TestOrderCreatedTopic”

3. Subscribe SQS เข้ากับ SNS (เชื่อมโทรโข่งกับคิว)

เรายังอยู่ในหน้าของ SNS นะ โดยเราจะสร้างให้ SQS ทั้ง 2 ตัวที่เราสร้างไว้ ให้มา subscribe ตัว SNS นี้จ้า (ตามรูปเลย)

เท่านี้ก้อจะจบแล้วนะ setup in AWS

2 SQS have been subscripted to a SNS

4. Coding in Python

• สร้าง project แล้วมาลุยกันเลย

# Create venv
python3 -m venv venv
source venv/bin/activate

# Install boto3
pip install boto3

# Config aws cli (กรอก AK, SK, region name = ap-southeast-7)
aws configure 

• สร้าง publish_order.py > Script นี้จำลองว่าลูกค้ากดสั่งซื้อสำเร็จ แล้วระบบโยนข้อความเข้า SNS

import boto3
import json
import uuid
import config

def publish_order():
    sns = boto3.client(
        'sns',
        region_name=config.AWS_REGION,
        aws_access_key_id=config.AWS_ACCESS_KEY_ID,
        aws_secret_access_key=config.AWS_SECRET_ACCESS_KEY,
    )

    # จำลองข้อมูลออเดอร์
    order_data = {
        "order_id": str(uuid.uuid4()),
        "item": "Mechanical Keyboard",
        "quantity": 1,
        "customer_email": "q_electronics@hotmail.com"
    }

    # ส่งข้อความเข้า SNS
    response = sns.publish(
        TopicArn=config.SNS_TOPIC_ARN,
        Message=json.dumps(order_data),
        Subject="New Order Created by GR"
    )

    print(f"✅ สร้างออเดอร์สำเร็จ! Message ID: {response['MessageId']}")
    print(f"ข้อมูล: {order_data}")

if __name__ == '__main__':
    publish_order()

• สร้าง process_inventory.py > เป็นฝั่งประมวลผลนะ

import boto3
import json
import time
import config

def process_queue():
    sqs = boto3.client(
        'sqs',
        region_name=config.AWS_REGION,
        aws_access_key_id=config.AWS_ACCESS_KEY_ID,
        aws_secret_access_key=config.AWS_SECRET_ACCESS_KEY,
    )

    print("⏳ รอรับงานจาก Inventory Queue...")

    while True:
        # ดึงข้อความจากคิว (Long Polling รอ 10 วินาทีเพื่อประหยัดทรัพยากร)
        response = sqs.receive_message(
            QueueUrl=config.SQS_QUEUE_URL,
            MaxNumberOfMessages=1,
            WaitTimeSeconds=10 
        )

        messages = response.get('Messages', [])

        if not messages:
            continue

        for message in messages:
            # SNS จะส่ง Message มาในรูปแบบ JSON ซ้อน JSON ต้อง Extract ออกมา
            sns_body = json.loads(message['Body'])
            order_data = json.loads(sns_body['Message'])

            print(f"\n📦 กำลังตัดสต็อกสำหรับ Order ID: {order_data['order_id']}")
            print(f"สินค้า: {order_data['item']} จำนวน: {order_data['quantity']}")

            # จำลองการใช้เวลาประมวลผล
            time.sleep(2)
            print("✅ ตัดสต็อกเสร็จสิ้น!")

            # ต้องลบข้อความออกจากคิวเมื่อทำงานเสร็จ ไม่งั้นข้อความจะเด้งกลับมาทำซ้ำ
            sqs.delete_message(
                QueueUrl=config.SQS_QUEUE_URL,
                ReceiptHandle=message['ReceiptHandle']
            )
            print("🗑️ ลบข้อความออกจากคิวแล้ว")

if __name__ == '__main__':
    process_queue()

• สร้าง config.py

AWS_ACCESS_KEY_ID = 'AKxxxxxxxxxxxxxx'
AWS_SECRET_ACCESS_KEY = 'UNxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
AWS_REGION = 'ap-southeast-7'

SNS_TOPIC_ARN = 'arn:aws:sns:ap-southeast-7:154230581564:TestOrderCreatedTopic'
SQS_QUEUE_URL = 'https://sqs.ap-southeast-7.amazonaws.com/154230581564/TestInventoryQueue'Run in 2 terminals

• มา Run in terminal กันเลย โดยแยกเป็น 2 terminals แล้วรัน 2 คำสั่งนี้นะ

# Terminal 1
# จะค้าง เพื่อรอรับงานจาก Inventory Queue...เพื่อไป process ต่อจ้า
python process_inventory.py

# Terminal 2
# ส่งงานเข้าไปจ้า (เป็นการสร้าง order เข้าไปใน SNS นั่นแหละ)
python publish_order.py 

มาดูการทำงานกันดีกว่านะ (ตามรูปเลย เข้าใจง่ายฝุดๆล่ะ)

• มาสำรวจ console ของ SQS กันต่อนะ ว่าที่ยังขึ้น Messages availabe เป็น 3 เราจะเห็นอะไรได้บ้าง

เราเข้ามาดู message ที่ยังค้างใน queue ได้นะ

• มาดูการทำงานของ code กันดีกว่าคับ ว่ามีการเชื่อมโยง SNS, SQS ยังไง

publish_order.py    SNS Topic             SQS Queue                Consumer
     │                  │                     │
     │──── publish ────►│                     │
     │                  │── fan out ──► TestInventoryQueue ◄── process_inventory.py
     │                  │── fan out ──► TestEmailQueue     ◄── (❌ not implement yet)

การทำงานของ python ทั้งส่วน publish_order และ process_inventory

References:

• Gemini: https://weuse.uk/3XGR, https://weuse.uk/oN8n (ของแทร่)
• Path: /Users/atthana/Desktop/Private_Q/Python/sqs_sns
• https://github.com/atthana/sqs_sns_testing

If you think it’s useful for you, just clap your hands 👏 to be encouraged me.

GRASSROOT ENGINEER 😘