الگوهای ارتباطی بین سرویس‌ها

۳.۱ مقدمه

یکی از مهم‌ترین تصمیمات در طراحی میکروسرویس، انتخاب نحوه ارتباط بین سرویس‌هاست. در این فصل با تمام الگوهای ارتباطی آشنا می‌شویم و یاد می‌گیریم چه زمانی از کدام استفاده کنیم.

۳.۲ Sync در مقابل Async

قانون کلی انتخاب

۳.۳ REST API

REST (Representational State Transfer) رایج‌ترین روش ارتباط بین میکروسرویس‌هاست. سادگی، پشتیبانی همه‌جانبه و human-readable بودن آن را به انتخاب پیش‌فرض تبدیل کرده است.

اصول REST

• Stateless: هر درخواست مستقل است؛ سرور state کاربر را نگه نمی‌دارد
• Resource-based: URL ها به منابع اشاره می‌کنند، نه action ها
• HTTP Verbs: GET (خواندن)، POST (ایجاد)، PUT (به‌روزرسانی کامل)، PATCH (به‌روزرسانی جزئی)، DELETE (حذف)
• Cacheable: پاسخ‌ها می‌توانند cache شوند
• Layered: client نمی‌داند مستقیماً به سرور وصل است یا از طریق proxy

طراحی URL مناسب

پیاده‌سازی با FastAPI

# product_service/main.py
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

app = FastAPI(title="Product Service", version="1.0.0")

class ProductCreate(BaseModel):
    name: str
    description: str
    price: float
    category_id: int

class ProductResponse(BaseModel):
    id: int
    name: str
    description: str
    price: float
    category_id: int
    created_at: datetime

# In-memory برای مثال
products_db = {}
next_id = 1

@app.get("/api/v1/products", response_model=List[ProductResponse])
async def list_products(skip: int = 0, limit: int = 100):
    """لیست محصولات با pagination"""
    products = list(products_db.values())
    return products[skip:skip + limit]

@app.get("/api/v1/products/{product_id}", response_model=ProductResponse)
async def get_product(product_id: int):
    """دریافت یک محصول"""
    if product_id not in products_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="محصول یافت نشد"
        )
    return products_db[product_id]

@app.post("/api/v1/products", response_model=ProductResponse,
          status_code=status.HTTP_201_CREATED)
async def create_product(product: ProductCreate):
    """ایجاد محصول جدید"""
    global next_id
    new_product = {
        "id": next_id,
        **product.dict(),
        "created_at": datetime.utcnow()
    }
    products_db[next_id] = new_product
    next_id += 1
    return new_product

@app.delete("/api/v1/products/{product_id}",
            status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(product_id: int):
    """حذف محصول"""
    if product_id not in products_db:
        raise HTTPException(status_code=404, detail="محصول یافت نشد")
    del products_db[product_id]
    return None
    

فراخوانی از سرویس دیگر

# order_service/clients/product_client.py
import httpx
from typing import Optional
from circuitbreaker import circuit  # برای fault tolerance

PRODUCT_SERVICE_URL = "http://product-service:8002"

class ProductClient:
    def __init__(self, timeout: float = 5.0):
        self.client = httpx.AsyncClient(
            base_url=PRODUCT_SERVICE_URL,
            timeout=timeout
        )
    
    @circuit(failure_threshold=5, recovery_timeout=30)
    async def get_product(self, product_id: int) -> Optional[dict]:
        """دریافت اطلاعات محصول"""
        try:
            response = await self.client.get(
                f"/api/v1/products/{product_id}"
            )
            if response.status_code == 404:
                return None
            response.raise_for_status()
            return response.json()
        except httpx.TimeoutException:
            # log timeout
            raise
        except httpx.HTTPError as e:
            # log error
            raise
    
    async def close(self):
        await self.client.aclose()

# استفاده در سرویس Order
from .clients.product_client import ProductClient

async def create_order(items: list[dict]):
    product_client = ProductClient()
    try:
        # برای هر آیتم، اطلاعات محصول را بگیر
        for item in items:
            product = await product_client.get_product(item["product_id"])
            if not product:
                raise ValueError(f"Product {item['product_id']} not found")
            # ادامه منطق سفارش...
    finally:
        await product_client.close()
    

HTTP Status Codes صحیح

• 2xx Success: 200 (OK), 201 (Created), 204 (No Content)
• 3xx Redirect: 301 (Moved Permanently), 304 (Not Modified)
• 4xx Client Error: 400 (Bad Request), 401 (Unauthorized), 403 (Forbidden), 404 (Not Found), 409 (Conflict), 422 (Validation Error), 429 (Too Many Requests)
• 5xx Server Error: 500 (Internal Error), 502 (Bad Gateway), 503 (Service Unavailable), 504 (Gateway Timeout)

۳.۴ gRPC

gRPC یک framework RPC مدرن و با کارایی بالا است که توسط Google توسعه داده شده. از HTTP/2 و Protocol Buffers (protobuf) برای سرعت و کارایی استفاده می‌کند.

مقایسه با REST

تعریف Service با Protobuf

// product.proto
syntax = "proto3";

package product;

service ProductService {
    rpc GetProduct(GetProductRequest) returns (Product);
    rpc ListProducts(ListProductsRequest) returns (ListProductsResponse);
    rpc CreateProduct(CreateProductRequest) returns (Product);
    rpc StreamProducts(StreamRequest) returns (stream Product);
}

message Product {
    int32 id = 1;
    string name = 2;
    string description = 3;
    double price = 4;
    int32 category_id = 5;
    int64 created_at = 6;
}

message GetProductRequest {
    int32 id = 1;
}

message ListProductsRequest {
    int32 skip = 1;
    int32 limit = 2;
    optional int32 category_id = 3;
}

message ListProductsResponse {
    repeated Product products = 1;
    int32 total = 2;
}

message CreateProductRequest {
    string name = 1;
    string description = 2;
    double price = 3;
    int32 category_id = 4;
}

message StreamRequest {
    int32 batch_size = 1;
}
    

تولید کد Python

# نصب ابزارها
pip install grpcio grpcio-tools

# تولید کد از .proto
python -m grpc_tools.protoc 
    --python_out=. 
    --grpc_python_out=. 
    --proto_path=. 
    product.proto

# خروجی: product_pb2.py و product_pb2_grpc.py
    

پیاده‌سازی Server

# product_server.py
import grpc
from concurrent import futures
import product_pb2
import product_pb2_grpc

class ProductServicer(product_pb2_grpc.ProductServiceServicer):
    def __init__(self):
        self.products = {}  # in-memory
        self.next_id = 1
    
    def GetProduct(self, request, context):
        product = self.products.get(request.id)
        if not product:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details(f"Product {request.id} not found")
            return product_pb2.Product()
        return product
    
    def ListProducts(self, request, context):
        all_products = list(self.products.values())
        if request.HasField("category_id"):
            all_products = [p for p in all_products 
                           if p.category_id == request.category_id]
        
        sliced = all_products[request.skip:request.skip + request.limit]
        return product_pb2.ListProductsResponse(
            products=sliced,
            total=len(all_products)
        )
    
    def CreateProduct(self, request, context):
        product = product_pb2.Product(
            id=self.next_id,
            name=request.name,
            description=request.description,
            price=request.price,
            category_id=request.category_id,
            created_at=int(time.time())
        )
        self.products[self.next_id] = product
        self.next_id += 1
        return product
    
    def StreamProducts(self, request, context):
        """مثال streaming - محصولات را batch به batch می‌فرستد"""
        import time
        for product in self.products.values():
            yield product
            time.sleep(0.1)  # شبیه‌سازی پردازش

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    product_pb2_grpc.add_ProductServiceServicer_to_server(
        ProductServicer(), server
    )
    server.add_insecure_port("[::]:50051")
    server.start()
    print("gRPC server running on port 50051")
    server.wait_for_termination()

if __name__ == "__main__":
    serve()
    

پیاده‌سازی Client

# product_client.py
import grpc
import product_pb2
import product_pb2_grpc

def get_product(product_id: int):
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = product_pb2_grpc.ProductServiceStub(channel)
        try:
            request = product_pb2.GetProductRequest(id=product_id)
            response = stub.GetProduct(request, timeout=5.0)
            return {
                "id": response.id,
                "name": response.name,
                "price": response.price
            }
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.NOT_FOUND:
                return None
            raise

def stream_products():
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = product_pb2_grpc.ProductServiceStub(channel)
        request = product_pb2.StreamRequest(batch_size=10)
        for product in stub.StreamProducts(request):
            print(f"Received: {product.name}")
    

۴ نوع Streaming در gRPC

1. Unary: یک request، یک response (مثل REST)
2. Server Streaming: یک request، چندین response (مثل SSE)
3. Client Streaming: چندین request، یک response (آپلود فایل)
4. Bidirectional Streaming: چندین در هر دو جهت (chat، live data)

۳.۵ GraphQL

GraphQL یک query language برای API هاست که توسط Facebook توسعه داده شد. مزیت اصلی آن: client می‌گوید دقیقاً چه داده‌ای می‌خواهد و overfetching/underfetching را حل می‌کند.

کاربرد در میکروسرویس

معمولاً GraphQL به عنوان API Gateway یا BFF استفاده می‌شود تا داده‌های چند سرویس را یکپارچه کند.

[Client]
   │
   │ GraphQL Query
   ▼
[GraphQL Gateway]
   │
   ├── REST ──► User Service
   ├── REST ──► Product Service
   ├── gRPC ──► Order Service
   └── REST ──► Review Service
    

مثال با Strawberry (Python)

import strawberry
from typing import List, Optional
import httpx

@strawberry.type
class Product:
    id: int
    name: str
    price: float

@strawberry.type
class Order:
    id: int
    total: float
    products: List[Product]

@strawberry.type
class User:
    id: int
    name: str
    email: str
    orders: List[Order]

@strawberry.type
class Query:
    @strawberry.field
    async def user(self, id: int) -> Optional[User]:
        # فراخوانی موازی به چند سرویس
        async with httpx.AsyncClient() as client:
            # دریافت اطلاعات کاربر
            user_resp = await client.get(f"http://user-service/users/{id}")
            user_data = user_resp.json()
            
            # دریافت سفارش‌های کاربر
            orders_resp = await client.get(
                f"http://order-service/orders?user_id={id}"
            )
            orders_data = orders_resp.json()
            
            return User(
                id=user_data["id"],
                name=user_data["name"],
                email=user_data["email"],
                orders=[Order(**o) for o in orders_data]
            )

schema = strawberry.Schema(query=Query)

# با FastAPI
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter

app = FastAPI()
graphql_app = GraphQLRouter(schema)
app.include_router(graphql_app, prefix="/graphql")
    

Query از سمت Client

query GetUserDetails {
  user(id: 1) {
    name
    email
    orders {
      id
      total
      products {
        name
        price
      }
    }
  }
}
    

client دقیقاً همان فیلدهایی که می‌خواهد را دریافت می‌کند، نه بیشتر و نه کمتر.

۳.۶ Message Queue (RabbitMQ)

Message Queue یک سیستم async است که در آن یک سرویس (Producer) پیام را در صف می‌گذارد و سرویس دیگر (Consumer) آن را پردازش می‌کند.

مفاهیم RabbitMQ

• Producer: فرستنده پیام
• Queue: صف پیام‌ها
• Consumer: دریافت‌کننده پیام
• Exchange: مسیریاب پیام به صف‌های مناسب
• Binding: قاعده اتصال exchange به queue
• Routing Key: کلید مسیریابی

۴ نوع Exchange

1. Direct: پیام به queue با routing_key دقیقاً مطابق ارسال می‌شود
2. Fanout: پیام به همه queue های متصل ارسال می‌شود (broadcast)
3. Topic: پیام بر اساس pattern routing key ارسال می‌شود (مثل user.*.created)
4. Headers: پیام بر اساس header ها مسیریابی می‌شود

پیاده‌سازی Producer

# producer.py - Order Service
import pika
import json
from datetime import datetime

def get_connection():
    credentials = pika.PlainCredentials("admin", "admin")
    parameters = pika.ConnectionParameters(
        host="rabbitmq",
        port=5672,
        credentials=credentials,
        heartbeat=600,
        blocked_connection_timeout=300
    )
    return pika.BlockingConnection(parameters)

def publish_order_event(order_data: dict, event_type: str):
    """انتشار event مربوط به سفارش"""
    connection = get_connection()
    channel = connection.channel()
    
    # تعریف exchange (idempotent)
    channel.exchange_declare(
        exchange="orders",
        exchange_type="topic",
        durable=True  # بعد از restart حفظ می‌شود
    )
    
    # ساخت پیام
    message = {
        "event_type": event_type,
        "order_id": order_data["id"],
        "user_id": order_data["user_id"],
        "total": order_data["total"],
        "timestamp": datetime.utcnow().isoformat(),
        "data": order_data
    }
    
    # ارسال
    routing_key = f"order.{event_type}"  # مثل order.placed
    channel.basic_publish(
        exchange="orders",
        routing_key=routing_key,
        body=json.dumps(message),
        properties=pika.BasicProperties(
            delivery_mode=2,  # persistent
            content_type="application/json",
            message_id=str(order_data["id"]),
        )
    )
    
    connection.close()

# استفاده
order = {"id": 123, "user_id": 5, "total": 1500000}
publish_order_event(order, "placed")  # routing_key: order.placed
    

پیاده‌سازی Consumer

# consumer.py - Notification Service
import pika
import json

def callback(ch, method, properties, body):
    """پردازش پیام دریافتی"""
    try:
        message = json.loads(body)
        print(f"Received: {message['event_type']} for order {message['order_id']}")
        
        if message["event_type"] == "placed":
            send_order_confirmation_email(message["data"])
        elif message["event_type"] == "shipped":
            send_shipping_notification(message["data"])
        elif message["event_type"] == "cancelled":
            send_cancellation_email(message["data"])
        
        # تأیید پردازش (acknowledge)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"Error: {e}")
        # rejected - برای retry به DLQ می‌رود
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

def consume():
    connection = get_connection()
    channel = connection.channel()
    
    # تعریف exchange
    channel.exchange_declare(
        exchange="orders",
        exchange_type="topic",
        durable=True
    )
    
    # تعریف queue اختصاصی برای این consumer
    queue_name = "notification_orders_queue"
    channel.queue_declare(queue=queue_name, durable=True)
    
    # binding: همه order events رو دریافت کن
    channel.queue_bind(
        exchange="orders",
        queue=queue_name,
        routing_key="order.*"
    )
    
    # محدود کردن prefetch برای load balancing
    channel.basic_qos(prefetch_count=1)
    
    channel.basic_consume(
        queue=queue_name,
        on_message_callback=callback,
        auto_ack=False  # manual ack
    )
    
    print("Waiting for messages...")
    channel.start_consuming()

if __name__ == "__main__":
    consume()
    

الگوهای پیشرفته

Dead Letter Queue (DLQ)

پیام‌هایی که چندین بار پردازششان شکست خورده، به DLQ منتقل می‌شوند برای بررسی دستی.

# تعریف queue با DLQ
channel.queue_declare(
    queue="orders_queue",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "dlx",
        "x-dead-letter-routing-key": "orders_dlq",
        "x-max-retries": 3
    }
)

channel.exchange_declare(exchange="dlx", exchange_type="direct")
channel.queue_declare(queue="orders_dlq", durable=True)
channel.queue_bind(exchange="dlx", queue="orders_dlq", routing_key="orders_dlq")
    

Retry با Exponential Backoff

import time

def callback_with_retry(ch, method, properties, body):
    retry_count = properties.headers.get("x-retry-count", 0) if properties.headers else 0
    
    try:
        process_message(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        if retry_count < 3:
            # backoff: 1s, 2s, 4s
            delay = 2 ** retry_count
            time.sleep(delay)
            
            # republish با retry count
            ch.basic_publish(
                exchange="orders",
                routing_key=method.routing_key,
                body=body,
                properties=pika.BasicProperties(
                    headers={"x-retry-count": retry_count + 1}
                )
            )
            ch.basic_ack(delivery_tag=method.delivery_tag)
        else:
            # به DLQ منتقل کن
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
    

۳.۷ Celery — Task Queue برای Python

Celery یک task queue توزیع‌شده محبوب برای Python است که از RabbitMQ یا Redis به عنوان broker استفاده می‌کند. برای کارهای async، scheduled و background ایده‌آل است.

کاربردها

• ارسال ایمیل و SMS
• پردازش تصاویر و ویدیو
• Webhook delivery
• گزارش‌گیری و export
• کارهای دوره‌ای (cron jobs)
• ارتباط async بین میکروسرویس‌ها

پیکربندی پایه

# celery_app.py
from celery import Celery

app = Celery(
    "notification_service",
    broker="amqp://admin:admin@rabbitmq:5672//",
    backend="redis://redis:6379/0",  # برای ذخیره نتایج
    include=["tasks.email", "tasks.sms"]
)

# تنظیمات
app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Asia/Tehran",
    enable_utc=True,
    task_track_started=True,
    task_time_limit=300,  # ۵ دقیقه
    task_soft_time_limit=240,
    worker_prefetch_multiplier=4,
    task_acks_late=True,  # ack بعد از موفقیت
    task_reject_on_worker_lost=True,
    
    # Task routing
    task_routes={
        "tasks.email.*": {"queue": "email"},
        "tasks.sms.*": {"queue": "sms"},
    }
)
    

تعریف Task ها

# tasks/email.py
from celery import shared_task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(
    bind=True,
    autoretry_for=(Exception,),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
    max_retries=5
)
def send_order_confirmation(self, order_id: int, email: str):
    """ارسال ایمیل تأیید سفارش"""
    try:
        logger.info(f"Sending confirmation to {email} for order {order_id}")
        # کد ارسال ایمیل
        send_email(to=email, subject="تأیید سفارش", template="order_confirmation")
        return {"status": "sent", "order_id": order_id}
    except Exception as e:
        logger.error(f"Failed: {e}")
        raise self.retry(exc=e, countdown=2 ** self.request.retries)

@shared_task
def send_bulk_emails(emails: list, subject: str, body: str):
    """ارسال انبوه ایمیل"""
    results = []
    for email in emails:
        try:
            send_email(email, subject, body)
            results.append({"email": email, "status": "ok"})
        except Exception as e:
            results.append({"email": email, "status": "error", "error": str(e)})
    return results
    

فراخوانی Task ها

from tasks.email import send_order_confirmation

# اجرای async
result = send_order_confirmation.delay(order_id=123, email="user@example.com")

# با countdown - بعد از ۶۰ ثانیه
result = send_order_confirmation.apply_async(
    args=[123, "user@example.com"],
    countdown=60
)

# در زمان مشخص
from datetime import datetime, timedelta
result = send_order_confirmation.apply_async(
    args=[123, "user@example.com"],
    eta=datetime.utcnow() + timedelta(hours=1)
)

# بررسی نتیجه
print(result.id)        # task id
print(result.status)    # PENDING, SUCCESS, FAILURE
print(result.ready())   # bool
print(result.get(timeout=10))  # دریافت نتیجه (blocking)
    

Periodic Tasks (Beat Scheduler)

from celery.schedules import crontab

app.conf.beat_schedule = {
    "send-daily-report": {
        "task": "tasks.report.send_daily_report",
        "schedule": crontab(hour=8, minute=0),  # هر روز ساعت ۸
    },
    "cleanup-old-sessions": {
        "task": "tasks.maintenance.cleanup_sessions",
        "schedule": crontab(minute=0),  # هر ساعت
    },
    "sync-products": {
        "task": "tasks.sync.sync_products",
        "schedule": 30.0,  # هر ۳۰ ثانیه
    }
}
    

اجرای Worker

# Worker معمولی
celery -A celery_app worker --loglevel=info

# با concurrency و queue اختصاصی
celery -A celery_app worker -Q email --concurrency=4 --loglevel=info

# Beat scheduler
celery -A celery_app beat --loglevel=info

# Flower (monitoring UI)
celery -A celery_app flower --port=5555
# سپس http://localhost:5555
    

۳.۸ Apache Kafka — Event Streaming

Kafka یک سیستم event streaming توزیع‌شده با throughput بسیار بالاست که برای real-time data pipelines و event-driven architecture استفاده می‌شود.

تفاوت با RabbitMQ

مفاهیم کلیدی

• Topic: دسته پیام‌ها (مثل صف ولی پایدار)
• Partition: هر topic به چند partition تقسیم می‌شود برای parallelism
• Offset: شماره ترتیب هر پیام در partition
• Producer: فرستنده
• Consumer: خواننده
• Consumer Group: چند consumer که با هم یک topic را پردازش می‌کنند
• Broker: یک سرور Kafka
• Cluster: چند broker که با هم کار می‌کنند

پیاده‌سازی Producer در Python

# kafka_producer.py
from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=["kafka:9092"],
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    key_serializer=lambda k: str(k).encode("utf-8"),
    acks="all",  # منتظر تأیید همه replica ها
    retries=3,
    compression_type="gzip"
)

def publish_user_event(user_id: int, event_type: str, data: dict):
    event = {
        "user_id": user_id,
        "event_type": event_type,
        "timestamp": time.time(),
        "data": data
    }
    
    # کلید (key) برای تعیین partition - همه event های یک کاربر به یک partition
    future = producer.send(
        topic="user-events",
        key=user_id,
        value=event
    )
    
    # blocking تا تأیید
    try:
        record_metadata = future.get(timeout=10)
        print(f"Sent to {record_metadata.topic} "
              f"partition {record_metadata.partition} "
              f"offset {record_metadata.offset}")
    except Exception as e:
        print(f"Failed to send: {e}")

publish_user_event(123, "login", {"ip": "192.168.1.1"})
producer.flush()
    

پیاده‌سازی Consumer

# kafka_consumer.py
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "user-events",  # topic
    bootstrap_servers=["kafka:9092"],
    group_id="analytics-service",  # consumer group
    auto_offset_reset="earliest",  # یا 'latest'
    enable_auto_commit=False,  # manual commit برای کنترل دقیق
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    key_deserializer=lambda k: k.decode("utf-8") if k else None,
    max_poll_records=100
)

for message in consumer:
    try:
        print(f"Received: partition={message.partition}, "
              f"offset={message.offset}, "
              f"key={message.key}, "
              f"value={message.value}")
        
        # پردازش پیام
        process_event(message.value)
        
        # commit دستی
        consumer.commit()
    except Exception as e:
        print(f"Error processing: {e}")
        # ندیدن offset باعث reprocess می‌شود
    

۳.۹ کِی از کدام استفاده کنیم؟

۳.۱۰ بهترین تجربیات

1. ترکیب hybrid معمولاً بهترین است. از REST برای API عمومی، gRPC برای داخلی، و RabbitMQ/Kafka برای events استفاده کنید.
2. برای commands، async ترجیح بدهید. «ارسال ایمیل» نباید synchronous باشد.
3. برای queries، sync ترجیح بدهید. «دریافت لیست محصولات» باید سریع پاسخ دهد.
4. API ها را versioned کنید. /api/v1/، /api/v2/ یا header.
5. Circuit Breaker اضافه کنید. برای جلوگیری از cascading failure.
6. Timeout مناسب تعریف کنید. هیچ‌گاه بدون timeout request نزنید.
7. Idempotency تضمین کنید. پیام‌های async ممکن است چند بار delivered شوند.
8. Schema را evolution-friendly طراحی کنید. فیلدهای جدید اختیاری باشند.
9. برای پیام‌های مهم از persistent storage استفاده کنید. در RabbitMQ delivery_mode=2.
10. Distributed Tracing اضافه کنید. برای debugging در محیط‌های پیچیده ضروری است.
11. Dead Letter Queue را تنظیم کنید. پیام‌های ناتوان از پردازش گم نشوند.
12. Backward compatibility را حفظ کنید. هیچ‌گاه فیلد قدیمی را حذف نکنید — deprecate کنید.

۳.۱۱ خلاصه فصل