ربط الخدمات: SQS وSNS وKinesis

إلى الحين أجزاء تطبيقنا تتصل ببعض مباشرة. هذا تمام لين يبدأ العرض وما تلحق خدمة الفواتير، وفجأة العميل يطالع دائرة تحميل بسبب خدمة ما سمع فيها.

هذا القسم عن التطبيقات الحديثة، ويبدأ بالفكرة اللي وراء أغلبها: افصل الخدمات عن بعض، عشان كل وحدة تشتغل بسرعتها، وتعطل وحدة ما يوقف الباقي.

هذي خريطة الجزء:

ليه تفصل الخدمات؟ #

الاتصال المباشر يعني الطرفين لازم يكونون شغالين وسريعين في نفس اللحظة. ولما تحط طابور بينهم تكسب ثلاث أشياء:

• العطل يبقى في مكانه. لو وقفت خدمة الفواتير، الرسائل تنتظر في الطابور لين ترجع.
• الضغط ينمتص. ألف طلب في دقيقة ما توصل لقاعدة البيانات دفعة وحدة.
• كل طرف يتوسع لحاله. تزيد عمّال الفواتير بدون ما تلمس المتجر.

SQS: الطابور #

SQS (Simple Queue Service) طابور مُدار بالكامل بدون سيرفرات تديرها. المنتج يرسل الرسائل، وتنتظر في الطابور، والمستهلك يسحبها ويشتغل عليها ثم يحذفها.

انتبه للاتجاه: المستهلكين هم اللي يطلبون الرسائل من الطابور، والطابور ما يدفعها لأحد.

آخر نقطتين تأثر على كودك. بما إن الرسالة ممكن توصل مرتين، خل معالجتها مرتين ما تضر. رقم طلب تتأكد منه قبل الخصم يكفي.

وأي شي أكبر من رسالة، خزّن الملف في S3 وأرسل مساره في الرسالة.

مهلة الإخفاء (Visibility Timeout) #

وش يمنع مستهلكين ياخذون نفس الرسالة؟ لما يستلمها واحد، يخفيها SQS عن الباقين لفترة اسمها Visibility Timeout. ولازم المستهلك يحذفها قبل ما تخلص الفترة.

هذي الآلية هي اللي تحميك من ضياع الشغل: لو طاح السيرفر في نص المعالجة، الرسالة ترجع لحالها.

اختر الفترة حسب وقت معالجتك. هي 30 ثانية افتراضياً، وتوصل إلى 12 ساعة:

• قصيرة زيادة: الرسالة ترجع تظهر وهي لسا تحت المعالجة، فتنعالج مرتين.
• طويلة زيادة: لو فشل مستهلك، الرسالة تنتظر طويل قبل ما أحد يعيد المحاولة.

ولو شغلة أخذت وقت أطول من المعتاد، المستهلك يقدر يمدد الفترة لهذي الرسالة بـ ChangeMessageVisibility.

الانتظار الطويل (Long Polling) #

افتراضياً، لما تسأل طابور فاضي يرجع لك رد فاضي فوراً، فمستهلكينك يملون فاتورتك بطلبات ما لها داعي. مع Long Polling الطلب ينتظر حتى 20 ثانية لين تجي رسالة. يعني طلبات أقل، وتكلفة أقل، ووصول أسرع، فشغّله دايماً.

طابور الرسائل الفاشلة (DLQ) #

طيب ورسالة تفشل كل مرة بسبب خطأ في الكود أو بيانات غلط؟ بترجع للأبد. اربط Dead-Letter Queue: بعد عدد محاولات تحدده، الرسالة تنتقل له، فتفحصها بدون ما تعطل شي. وبعد ما تصلح المشكلة، تقدر تعيدها للطابور الأصلي بـ Redrive.

وخل مدة الحفظ في طابور الرسائل الفاشلة أطول من الأصلي، عشان ما تنتهي الرسائل قبل ما تشوفها.

طوابير FIFO #

لما يكون الترتيب مهم، مثل عمليات على نفس الحساب، استخدم طابور FIFO. ولازم اسمه ينتهي بـ .fifo.

الترتيب يكون داخل مجموعة الرسائل (Message Group)، فلو خليت مجموعة لكل عميل، رسائل العملاء المختلفين تنعالج بالتوازي. ولو احتجت سعة أكبر، وضع السعة العالية يرفع الحد كثير، والرقم يختلف حسب المنطقة.

التوسع مع الطابور #

نمط كلاسيكي: مجموعة Auto Scaling من العمّال تكبر وتصغر حسب طول الطابور. مقياس ApproximateNumberOfMessagesVisible يقول لك كم رسالة تنتظر، فلما يطول الطابور يزيد العمّال.

وهذا يحمي قاعدة بياناتك كمان. بدل ما تضرب موجة طلبات RDS مباشرة، تدخل في طابور، والعمّال يكتبونها بسرعة تتحملها القاعدة. وهذا يناسب الشغل اللي ما يحتاج جواب فوري فقط، فالعميل يوصله "استلمنا طلبك"، مو "اكتمل طلبك".

الأمان #

الرسائل مشفرة أثناء النقل بـ HTTPS، وفي التخزين بـ SSE-SQS افتراضياً أو بمفتاح KMS خاص فيك. ومين يوصل للطابور تحدده سياسات IAM، مع سياسة وصول على الطابور نفسه، وتحتاجها لحسابات ثانية أو لخدمات مثل SNS وS3 عشان ترسل له. وبنكتب وحدة في التجربة العملية.

SNS: رسالة وحدة للكل #

الطابور يناسب شغلة يسويها طرف واحد. بس لما يطلب عميل، خدمات الشحن والفواتير والتحليل كلها تبي تدري. وما تبي تطبيقك يعرف كل وحدة فيها، ويتغير كل ما أضفت وحدة.

SNS (Simple Notification Service) يشتغل بـ النشر والاشتراك: تطبيقك ينشر رسالة وحدة في موضوع (Topic)، وSNS يوصلها لكل المشتركين فوراً. والموضوع العادي يتحمل حتى 12.5 مليون اشتراك.

وعلى عكس الطابور، SNS ما يخزن الرسائل. لو كان مشترك واقف يعيد المحاولة لفترة، بس هو مو مكان تنتظر فيه الرسائل.

وحجم الرسالة 256 KiB افتراضياً، وتقدر ترفعه على الموضوع إلى 1 MiB، بس هذا يحصره في 100 اشتراك ولمشتركي SQS وLambda وFirehose فقط.

Fan-out: SNS مع SQS #

أهم نمط هنا يجمع الخدمتين: SNS ينشر الرسالة، وطابور لكل خدمة يحفظها لين تكون الخدمة جاهزة.

ليه هذا الجمع؟

• ما يضيع شي: لو وقفت خدمة الشحن، رسائلها تنتظر في طابورها.
• كل خدمة لها سرعتها، ومحاولاتها، وطابور رسائلها الفاشلة.
• إضافة خدمة سهلة: طابور جديد يشترك في الموضوع، بدون تغيير الناشر.

ونفس الشي مع أحداث S3: S3 ما يقبل قاعدتين ببادئات متداخلة لنفس نوع الحدث. فعشان توصل لأكثر من وجهة، أرسل الحدث لموضوع SNS واحد ووزعه من هناك.

فلترة الرسائل #

افتراضياً كل مشترك يستلم كل شي. سياسة الفلترة على الاشتراك تخلي طابور الشحن يستلم اللي يخصه فقط:

{ "type": ["physical"] }

الرسائل اللي type فيها digital ما توصله، وما تدفع على معالجتها.

مواضيع FIFO #

عشان الترتيب في الـ Fan-out، SNS عنده مواضيع FIFO، وأسماؤها تنتهي بـ .fifo برضو، مع ترتيب وحذف للتكرار مثل طوابير FIFO. ومحدودة بـ 100 اشتراك للموضوع، وتناسب حالات مثل إرسال نفس تسلسل التحديثات لأكثر من طابور.

Kinesis Data Streams: تيار يبقى #

تخيل مليون حساس يرسلون قراءات كل ثانية، وثلاث أطراف تبي تقرأها: واحد للتنبيهات الفورية، وواحد للوحات المتابعة، وواحد للأرشفة. الطابور ما يناسب هنا، لأن الرسالة تروح بعد معالجتها.

Kinesis Data Streams تيار: السجلات تبقى فيه طول مدة الحفظ سواء انقرأت أو لا، وكل مستهلك يقرأ من موضعه، فيقدر يرجع ويعيد القراءة.

• Shards: التيار ينقسم إلى أجزاء. في السعة المحددة، كل واحد ياخذ 1 MB في الثانية كتابة و2 MB في الثانية قراءة.
• مفتاح التقسيم: يحدد أي Shard يروح له السجل. السجلات بنفس المفتاح تبقى مرتبة، فرقم الحساس مفتاح ممتاز.
• مدة الحفظ: 24 ساعة افتراضياً، وتوصل إلى 365 يوم.
• حجم السجل: 1 MiB افتراضياً، وتقدر ترفعه إلى 10 MiB لكل تيار.

أوضاع السعة #

وسقف On-demand يختلف حسب المنطقة: حتى 10 GB في الثانية كتابة في أكبر المناطق، و200 MB في الثانية في غيرها.

وللفيديو خدمة خاصة: Kinesis Video Streams تستقبل فيديو مباشر من أجهزة مثل الكاميرات، وتحفظه، وتخليك تشاهده أو تعالجه، مباشرة أو بعدين.

Data Firehose: للتخزين بدون كود #

كثير من الأحيان كل اللي تبيه إن التيار ينزل في مكان. Amazon Data Firehose (اسمه القديم Kinesis Data Firehose) ياخذ البيانات، ويجمعها، ويوصلها بدون كود ولا سيرفرات:

• المصادر: تطبيقك مباشرة، أو تيار Kinesis، أو غيرها مثل SNS وCloudWatch Logs.
• المعالجة: يقدر يحوّل السجلات بدالة Lambda لو تبي، ويحولها إلى Parquet.
• الوجهات: S3 وRedshift وOpenSearch وSnowflake وجداول Apache Iceberg، أو رابط HTTP.
• التجميع: يجمع البيانات لين توصل حجم أو مدة، مثل 5 MiB أو 5 دقايق لـ S3، واللي يجي أول يطلق التوصيل. فهو يناسب التحميل شبه الفوري، مو رد الفعل اللحظي.

أيها أختار؟ #

قاعدة سريعة: شغلة؟ SQS. حدث يهم كثيرين؟ SNS. تيار بيانات يقرأه كثيرين ويعيدون قراءته؟ Kinesis.

وفيه كمان EventBridge يوزع الأحداث بقواعد من خدمات AWS وتطبيقات ثانية، وبنشرحه في الجزء الخامس والعشرين.

Amazon MQ: للي موجود أصلاً #

SQS وSNS خدمات AWS الخاصة بواجهاتها الخاصة. لو تنقل تطبيق مبني على بروتوكولات مفتوحة مثل AMQP أو MQTT أو STOMP، ما تبي تعيد كتابته. Amazon MQ يشغّل لك وسيط معروف كخدمة مُدارة، بمحركين:

ولاحظ إنه يشتغل على وسطاء تختار حجمهم، فما يتوسع بلا حدود مثل SQS وSNS. وللتطبيق الجديد، ابدأ بهذيك الخدمتين.

عملي: Fan-out مع فلتر #

بنبني تصميم الرسم بطابورين. من CloudShell، أنشئ الموضوع والطوابير:

TOPIC_ARN=$(aws sns create-topic --name orders \
  --query TopicArn --output text)

SHIP_URL=$(aws sqs create-queue --queue-name shipping \
  --query QueueUrl --output text)
BILL_URL=$(aws sqs create-queue --queue-name billing \
  --query QueueUrl --output text)

SHIP_ARN=$(aws sqs get-queue-attributes --queue-url $SHIP_URL \
  --attribute-names QueueArn --query Attributes.QueueArn --output text)
BILL_ARN=$(aws sqs get-queue-attributes --queue-url $BILL_URL \
  --attribute-names QueueArn --query Attributes.QueueArn --output text)

الطوابير خاصة، فاسمح لهذا الموضوع فقط يرسل لها:

allow_topic() {
  POLICY=$(jq -nc --arg q "$2" --arg t "$TOPIC_ARN" '{
    Version: "2012-10-17",
    Statement: [{
      Effect: "Allow",
      Principal: {Service: "sns.amazonaws.com"},
      Action: "sqs:SendMessage",
      Resource: $q,
      Condition: {ArnEquals: {"aws:SourceArn": $t}}
    }]
  }')
  aws sqs set-queue-attributes --queue-url "$1" \
    --attributes "$(jq -nc --arg p "$POLICY" '{Policy: $p}')"
}

allow_topic $SHIP_URL $SHIP_ARN
allow_topic $BILL_URL $BILL_ARN

اشترك بالطوابير في الموضوع: الفواتير تاخذ كل شي، والشحن ياخذ الطلبات المادية فقط:

aws sns subscribe --topic-arn $TOPIC_ARN \
  --protocol sqs --notification-endpoint $BILL_ARN \
  --attributes RawMessageDelivery=true

aws sns subscribe --topic-arn $TOPIC_ARN \
  --protocol sqs --notification-endpoint $SHIP_ARN \
  --attributes '{"RawMessageDelivery":"true","FilterPolicy":"{\"type\":[\"physical\"]}"}'

انشر طلبين، واحد مادي وواحد رقمي:

aws sns publish --topic-arn $TOPIC_ARN --message "order 1: a book" \
  --message-attributes '{"type":{"DataType":"String","StringValue":"physical"}}'

aws sns publish --topic-arn $TOPIC_ARN --message "order 2: an e-book" \
  --message-attributes '{"type":{"DataType":"String","StringValue":"digital"}}'

عطه كم ثانية، ثم عد الرسائل في كل طابور:

for url in $BILL_URL $SHIP_URL; do
  aws sqs get-queue-attributes --queue-url $url \
    --attribute-names ApproximateNumberOfMessages \
    --query Attributes.ApproximateNumberOfMessages --output text
done

بتشوف 2 للفواتير و1 للشحن، يعني الفلتر اشتغل. الحين جرب مهلة الإخفاء. استلم رسالة الشحن، ثم حاول مرة ثانية على طول:

aws sqs receive-message --queue-url $SHIP_URL --wait-time-seconds 5 \
  --query "Messages[0].Body" --output text

aws sqs receive-message --queue-url $SHIP_URL --wait-time-seconds 5 \
  --query "Messages[0].Body" --output text

الأول يطبع order 1: a book، والثاني ينتظر خمس ثواني ويطبع None، لأن الرسالة مخفية. وما حذفناها، فانتظر 30 ثانية واستلمها مرة ثانية، وهالمرة احذفها عشان ينتهي الشغل:

HANDLE=$(aws sqs receive-message --queue-url $SHIP_URL --wait-time-seconds 20 \
  --query "Messages[0].ReceiptHandle" --output text)

aws sqs delete-message --queue-url $SHIP_URL --receipt-handle "$HANDLE"

الحين الرسالة راحت نهائياً.

aws sns delete-topic --topic-arn $TOPIC_ARN
aws sqs delete-queue --queue-url $SHIP_URL
aws sqs delete-queue --queue-url $BILL_URL

الخلاصة #

• فصل الخدمات يحصر الأعطال، ويمتص الضغط، ويخلي كل طرف يتوسع لحاله.
• SQS طابور: الرسالة تروح لمستهلك واحد، ولازم يحذفها قبل نهاية مهلة الإخفاء.
• أضف دايماً Long Polling وDLQ، وخل المعالجة آمنة لو تكررت.
• SNS يوصل رسالة وحدة للكل، وFan-out مع SQS يضمن ما تفوت أحد.
• Kinesis Data Streams يحفظ البيانات لأكثر من قارئ يعيد قراءتها، وFirehose يوصلها للتخزين بدون كود.

اختبر نفسك #

قبل ما تكمل، جرّب هذي الأسئلة عشان تتأكد إنك فهمت الجزء. كل سؤال موقف عملي، وبعد ما تجاوب يطلع لك الشرح.

الخدمات الحين مفصولة، بس وين يشتغل الكود نفسه؟ إلى الآن جوابنا سيرفر EC2 بكل إعداداته. في الجزء العشرين نغلف التطبيقات في حاويات ونشغلها بـ ECS وFargate وEKS.