Asynchronous Communication

Introduction

Asynchronous communication fundamentally changes how services interact. Instead of waiting for responses, services publish messages and move on. Through building event-driven systems, I've learned that async patterns solve coupling problems but introduce new challenges around eventual consistency, ordering, and debugging.

This article covers message queues, event-driven patterns, and practical implementations with RabbitMQ and Celery.

Why Asynchronous Communication?

Aspect	Synchronous	Asynchronous
Coupling	Tight	Loose
Availability	Dependent on all services	Tolerates service failures
Latency	Cumulative	Fire-and-forget
Consistency	Immediate	Eventual
Complexity	Lower	Higher

Message Queue Fundamentals

Core Concepts

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Callable, Any
import json
import uuid


@dataclass
class Message:
    """Base message structure."""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    type: str = ""
    payload: dict = field(default_factory=dict)
    metadata: dict = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.utcnow)
    
    def to_json(self) -> str:
        return json.dumps({
            "id": self.id,
            "type": self.type,
            "payload": self.payload,
            "metadata": self.metadata,
            "timestamp": self.timestamp.isoformat(),
        })
    
    @classmethod
    def from_json(cls, data: str) -> "Message":
        parsed = json.loads(data)
        parsed["timestamp"] = datetime.fromisoformat(parsed["timestamp"])
        return cls(**parsed)


class MessageBroker(ABC):
    """Abstract message broker interface."""
    
    @abstractmethod
    async def publish(self, queue: str, message: Message) -> None: ...
    
    @abstractmethod
    async def subscribe(
        self, 
        queue: str, 
        handler: Callable[[Message], Any]
    ) -> None: ...
    
    @abstractmethod
    async def close(self) -> None: ...

RabbitMQ Implementation

import aio_pika
from aio_pika import Message as AMQPMessage, ExchangeType


class RabbitMQBroker(MessageBroker):
    """RabbitMQ message broker implementation."""
    
    def __init__(self, url: str = "amqp://guest:guest@localhost/"):
        self.url = url
        self.connection = None
        self.channel = None
    
    async def connect(self):
        self.connection = await aio_pika.connect_robust(self.url)
        self.channel = await self.connection.channel()
        await self.channel.set_qos(prefetch_count=10)
    
    async def publish(self, queue: str, message: Message) -> None:
        if not self.channel:
            await self.connect()
        
        # Declare queue
        await self.channel.declare_queue(queue, durable=True)
        
        # Publish message
        await self.channel.default_exchange.publish(
            AMQPMessage(
                body=message.to_json().encode(),
                delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
                message_id=message.id,
                timestamp=message.timestamp,
                headers=message.metadata,
            ),
            routing_key=queue,
        )
    
    async def subscribe(
        self, 
        queue: str, 
        handler: Callable[[Message], Any]
    ) -> None:
        if not self.channel:
            await self.connect()
        
        # Declare queue
        queue_obj = await self.channel.declare_queue(queue, durable=True)
        
        async def process_message(amqp_message: aio_pika.IncomingMessage):
            async with amqp_message.process():
                message = Message.from_json(amqp_message.body.decode())
                await handler(message)
        
        await queue_obj.consume(process_message)
    
    async def close(self) -> None:
        if self.connection:
            await self.connection.close()


# Usage
async def main():
    broker = RabbitMQBroker()
    await broker.connect()
    
    # Publish
    await broker.publish(
        "orders",
        Message(type="order.created", payload={"order_id": "123"}),
    )
    
    # Subscribe
    async def handle_order(message: Message):
        print(f"Received: {message.payload}")
    
    await broker.subscribe("orders", handle_order)

Exchange Types in RabbitMQ

class RabbitMQExchangeBroker:
    """RabbitMQ with different exchange types."""
    
    async def setup_fanout_exchange(self, exchange_name: str):
        """Fanout: Broadcast to all bound queues."""
        exchange = await self.channel.declare_exchange(
            exchange_name,
            ExchangeType.FANOUT,
            durable=True,
        )
        return exchange
    
    async def setup_direct_exchange(self, exchange_name: str):
        """Direct: Route by exact routing key match."""
        exchange = await self.channel.declare_exchange(
            exchange_name,
            ExchangeType.DIRECT,
            durable=True,
        )
        return exchange
    
    async def setup_topic_exchange(self, exchange_name: str):
        """Topic: Route by routing key pattern."""
        exchange = await self.channel.declare_exchange(
            exchange_name,
            ExchangeType.TOPIC,
            durable=True,
        )
        return exchange
    
    async def publish_event(
        self, 
        exchange_name: str, 
        routing_key: str,
        message: Message
    ):
        """Publish to exchange with routing key."""
        exchange = await self.channel.get_exchange(exchange_name)
        await exchange.publish(
            AMQPMessage(body=message.to_json().encode()),
            routing_key=routing_key,
        )
    
    async def subscribe_with_pattern(
        self,
        exchange_name: str,
        routing_pattern: str,  # e.g., "order.*" or "order.#"
        handler: Callable,
    ):
        """Subscribe using routing pattern (topic exchange)."""
        exchange = await self.channel.get_exchange(exchange_name)
        queue = await self.channel.declare_queue(exclusive=True)
        await queue.bind(exchange, routing_pattern)
        await queue.consume(handler)


# Topic exchange patterns:
# order.created -> matches "order.*", "order.#", "#"
# order.item.added -> matches "order.#", "#", but NOT "order.*"

Event-Driven Architecture

Domain Events

from dataclasses import dataclass
from datetime import datetime
from typing import Generic, TypeVar
import uuid

T = TypeVar("T")


@dataclass
class DomainEvent(Generic[T]):
    """Base class for domain events."""
    event_id: str
    event_type: str
    aggregate_id: str
    aggregate_type: str
    data: T
    occurred_at: datetime
    version: int = 1
    
    @classmethod
    def create(
        cls, 
        event_type: str, 
        aggregate_id: str,
        aggregate_type: str,
        data: T
    ) -> "DomainEvent[T]":
        return cls(
            event_id=str(uuid.uuid4()),
            event_type=event_type,
            aggregate_id=aggregate_id,
            aggregate_type=aggregate_type,
            data=data,
            occurred_at=datetime.utcnow(),
        )


# Specific events
@dataclass
class OrderCreatedData:
    customer_id: str
    items: list[dict]
    total_amount: float


@dataclass
class OrderCancelledData:
    reason: str
    cancelled_by: str


@dataclass
class PaymentProcessedData:
    payment_id: str
    amount: float
    method: str


# Creating events
order_created = DomainEvent.create(
    event_type="order.created",
    aggregate_id="order-123",
    aggregate_type="Order",
    data=OrderCreatedData(
        customer_id="customer-456",
        items=[{"product_id": "prod-1", "quantity": 2}],
        total_amount=99.99,
    ),
)

Event Publisher

from abc import ABC, abstractmethod
from typing import Callable


class EventPublisher(ABC):
    """Abstract event publisher."""
    
    @abstractmethod
    async def publish(self, event: DomainEvent) -> None: ...


class RabbitMQEventPublisher(EventPublisher):
    """Publish events to RabbitMQ topic exchange."""
    
    def __init__(self, broker: RabbitMQBroker, exchange_name: str = "events"):
        self.broker = broker
        self.exchange_name = exchange_name
    
    async def setup(self):
        """Create topic exchange for events."""
        await self.broker.connect()
        self.exchange = await self.broker.channel.declare_exchange(
            self.exchange_name,
            ExchangeType.TOPIC,
            durable=True,
        )
    
    async def publish(self, event: DomainEvent) -> None:
        """Publish event with routing key based on event type."""
        message = AMQPMessage(
            body=json.dumps({
                "event_id": event.event_id,
                "event_type": event.event_type,
                "aggregate_id": event.aggregate_id,
                "aggregate_type": event.aggregate_type,
                "data": event.data.__dict__ if hasattr(event.data, "__dict__") else event.data,
                "occurred_at": event.occurred_at.isoformat(),
                "version": event.version,
            }).encode(),
            message_id=event.event_id,
            timestamp=event.occurred_at,
        )
        
        # Use event_type as routing key
        await self.exchange.publish(message, routing_key=event.event_type)


# Usage in Order Service
class OrderService:
    def __init__(self, repository, event_publisher: EventPublisher):
        self.repository = repository
        self.event_publisher = event_publisher
    
    async def create_order(self, customer_id: str, items: list) -> Order:
        # Create order
        order = Order(customer_id=customer_id, items=items)
        await self.repository.save(order)
        
        # Publish event
        event = DomainEvent.create(
            event_type="order.created",
            aggregate_id=order.id,
            aggregate_type="Order",
            data=OrderCreatedData(
                customer_id=customer_id,
                items=items,
                total_amount=order.total,
            ),
        )
        await self.event_publisher.publish(event)
        
        return order

Event Consumer

class EventConsumer:
    """Subscribe to and handle events."""
    
    def __init__(self, broker: RabbitMQBroker, exchange_name: str = "events"):
        self.broker = broker
        self.exchange_name = exchange_name
        self.handlers: dict[str, list[Callable]] = {}
    
    def on(self, event_pattern: str):
        """Decorator to register event handler."""
        def decorator(handler: Callable):
            if event_pattern not in self.handlers:
                self.handlers[event_pattern] = []
            self.handlers[event_pattern].append(handler)
            return handler
        return decorator
    
    async def start(self, service_name: str):
        """Start consuming events."""
        await self.broker.connect()
        
        exchange = await self.broker.channel.declare_exchange(
            self.exchange_name,
            ExchangeType.TOPIC,
            durable=True,
        )
        
        for pattern in self.handlers:
            # Create queue per service per pattern
            queue_name = f"{service_name}.{pattern.replace('.', '_')}"
            queue = await self.broker.channel.declare_queue(
                queue_name,
                durable=True,
            )
            await queue.bind(exchange, pattern)
            
            async def process(message: aio_pika.IncomingMessage, pattern=pattern):
                async with message.process():
                    event_data = json.loads(message.body.decode())
                    for handler in self.handlers[pattern]:
                        await handler(event_data)
            
            await queue.consume(process)


# Usage in Inventory Service
consumer = EventConsumer(broker)


@consumer.on("order.created")
async def handle_order_created(event: dict):
    """Reserve inventory when order is created."""
    order_id = event["aggregate_id"]
    items = event["data"]["items"]
    
    for item in items:
        await inventory_service.reserve(
            product_id=item["product_id"],
            quantity=item["quantity"],
            order_id=order_id,
        )


@consumer.on("order.cancelled")
async def handle_order_cancelled(event: dict):
    """Release inventory when order is cancelled."""
    order_id = event["aggregate_id"]
    await inventory_service.release_reservation(order_id)


# Start consuming
await consumer.start("inventory-service")

Task Queues with Celery

Celery Setup

# celeryconfig.py
broker_url = "amqp://guest:guest@localhost:5672//"
result_backend = "redis://localhost:6379/0"

task_serializer = "json"
result_serializer = "json"
accept_content = ["json"]
timezone = "UTC"
enable_utc = True

# Task routing
task_routes = {
    "tasks.email.*": {"queue": "email"},
    "tasks.payment.*": {"queue": "payment"},
    "tasks.notification.*": {"queue": "notification"},
}

# Retry settings
task_annotations = {
    "tasks.payment.process_payment": {
        "rate_limit": "10/m",
        "max_retries": 3,
    },
}

# tasks.py
from celery import Celery, Task
from celery.exceptions import MaxRetriesExceededError
import logging

app = Celery("tasks")
app.config_from_object("celeryconfig")

logger = logging.getLogger(__name__)


class BaseTask(Task):
    """Base task with error handling."""
    
    autoretry_for = (Exception,)
    retry_backoff = True
    retry_backoff_max = 600
    retry_jitter = True
    
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        logger.error(f"Task {self.name}[{task_id}] failed: {exc}")
    
    def on_success(self, retval, task_id, args, kwargs):
        logger.info(f"Task {self.name}[{task_id}] succeeded")


@app.task(bind=True, base=BaseTask, max_retries=3)
def process_payment(self, order_id: str, amount: float, method: str):
    """Process payment asynchronously."""
    try:
        # Simulate payment processing
        result = payment_gateway.charge(
            order_id=order_id,
            amount=amount,
            method=method,
        )
        
        if not result.success:
            raise PaymentFailedError(result.error)
        
        return {"payment_id": result.payment_id, "status": "completed"}
    
    except TransientError as e:
        # Retry on transient errors
        raise self.retry(exc=e, countdown=60)
    
    except PaymentFailedError as e:
        # Don't retry on business errors
        return {"status": "failed", "error": str(e)}


@app.task(bind=True, base=BaseTask)
def send_notification(self, user_id: str, message: str, channel: str = "email"):
    """Send notification to user."""
    if channel == "email":
        send_email(user_id, message)
    elif channel == "sms":
        send_sms(user_id, message)
    elif channel == "push":
        send_push(user_id, message)


@app.task(bind=True, base=BaseTask)
def process_order_async(self, order_data: dict):
    """Process order with multiple steps."""
    order_id = order_data["order_id"]
    
    # Chain multiple tasks
    from celery import chain
    
    workflow = chain(
        validate_inventory.s(order_data),
        process_payment.s(order_data),
        update_inventory.s(order_data),
        send_confirmation.s(order_data),
    )
    
    return workflow.apply_async()


# Calling tasks
from tasks import process_payment, send_notification

# Async call
result = process_payment.delay(
    order_id="order-123",
    amount=99.99,
    method="credit_card",
)

# Get result (if needed)
payment_result = result.get(timeout=30)

# Fire and forget
send_notification.delay(
    user_id="user-456",
    message="Your order has been shipped!",
    channel="email",
)

Task Patterns

from celery import group, chain, chord


# Chain: Sequential execution
workflow = chain(
    validate_order.s(order_id),
    process_payment.s(),
    send_confirmation.s(),
)
result = workflow.apply_async()


# Group: Parallel execution
notifications = group([
    send_email.s(user_id, "Order confirmed"),
    send_sms.s(user_id, "Order confirmed"),
    send_push.s(user_id, "Order confirmed"),
])
result = notifications.apply_async()


# Chord: Parallel with callback
from celery import chord

# Process all items, then send summary
item_tasks = group([
    process_item.s(item) 
    for item in order.items
])
workflow = chord(item_tasks)(send_order_summary.s(order_id))


# Periodic tasks
from celery.schedules import crontab

app.conf.beat_schedule = {
    "cleanup-expired-sessions": {
        "task": "tasks.cleanup_sessions",
        "schedule": crontab(hour=2, minute=0),  # Daily at 2 AM
    },
    "sync-inventory": {
        "task": "tasks.sync_inventory",
        "schedule": 300.0,  # Every 5 minutes
    },
}

Eventual Consistency

Understanding Eventual Consistency

Handling Eventual Consistency

from enum import Enum
from dataclasses import dataclass


class OrderStatus(Enum):
    PENDING = "pending"
    INVENTORY_RESERVED = "inventory_reserved"
    PAYMENT_PROCESSED = "payment_processed"
    CONFIRMED = "confirmed"
    FAILED = "failed"


@dataclass
class OrderState:
    order_id: str
    status: OrderStatus
    inventory_reserved: bool = False
    payment_processed: bool = False
    
    def is_complete(self) -> bool:
        return self.inventory_reserved and self.payment_processed


class OrderSagaManager:
    """Manage order saga state."""
    
    def __init__(self, repository, event_publisher):
        self.repository = repository
        self.event_publisher = event_publisher
    
    async def handle_inventory_reserved(self, order_id: str):
        """Handle inventory reservation completion."""
        state = await self.repository.get(order_id)
        state.inventory_reserved = True
        
        if state.is_complete():
            state.status = OrderStatus.CONFIRMED
            await self.event_publisher.publish(
                DomainEvent.create(
                    "order.confirmed",
                    order_id,
                    "Order",
                    {"status": "confirmed"},
                )
            )
        
        await self.repository.save(state)
    
    async def handle_payment_processed(self, order_id: str):
        """Handle payment completion."""
        state = await self.repository.get(order_id)
        state.payment_processed = True
        
        if state.is_complete():
            state.status = OrderStatus.CONFIRMED
            await self.event_publisher.publish(
                DomainEvent.create(
                    "order.confirmed",
                    order_id,
                    "Order",
                    {"status": "confirmed"},
                )
            )
        
        await self.repository.save(state)
    
    async def handle_failure(self, order_id: str, reason: str):
        """Handle saga failure - compensate."""
        state = await self.repository.get(order_id)
        state.status = OrderStatus.FAILED
        
        # Publish compensation events
        if state.inventory_reserved:
            await self.event_publisher.publish(
                DomainEvent.create(
                    "inventory.release_requested",
                    order_id,
                    "Order",
                    {"reason": reason},
                )
            )
        
        if state.payment_processed:
            await self.event_publisher.publish(
                DomainEvent.create(
                    "payment.refund_requested",
                    order_id,
                    "Order",
                    {"reason": reason},
                )
            )
        
        await self.repository.save(state)

Idempotency

from functools import wraps
import hashlib


class IdempotencyStore:
    """Store for tracking processed messages."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 86400  # 24 hours
    
    async def is_processed(self, idempotency_key: str) -> bool:
        return await self.redis.exists(f"idempotency:{idempotency_key}")
    
    async def mark_processed(self, idempotency_key: str, result: str):
        await self.redis.setex(
            f"idempotency:{idempotency_key}",
            self.ttl,
            result,
        )
    
    async def get_result(self, idempotency_key: str) -> str | None:
        return await self.redis.get(f"idempotency:{idempotency_key}")


def idempotent(key_func):
    """Decorator to make message handler idempotent."""
    def decorator(handler):
        @wraps(handler)
        async def wrapper(self, message: Message, *args, **kwargs):
            key = key_func(message)
            
            # Check if already processed
            if await self.idempotency_store.is_processed(key):
                cached_result = await self.idempotency_store.get_result(key)
                return json.loads(cached_result) if cached_result else None
            
            # Process message
            result = await handler(self, message, *args, **kwargs)
            
            # Mark as processed
            await self.idempotency_store.mark_processed(
                key,
                json.dumps(result) if result else "",
            )
            
            return result
        return wrapper
    return decorator


class PaymentEventHandler:
    def __init__(self, idempotency_store: IdempotencyStore):
        self.idempotency_store = idempotency_store
    
    @idempotent(key_func=lambda m: m.id)
    async def handle_payment_request(self, message: Message):
        """Process payment - idempotent."""
        order_id = message.payload["order_id"]
        amount = message.payload["amount"]
        
        # Process payment only once
        result = await self.process_payment(order_id, amount)
        return result

Message Ordering and Partitioning

class PartitionedEventPublisher:
    """Publish events with ordering guarantees within partitions."""
    
    def __init__(self, broker: RabbitMQBroker, num_partitions: int = 10):
        self.broker = broker
        self.num_partitions = num_partitions
    
    def _get_partition(self, partition_key: str) -> int:
        """Consistent hashing for partition assignment."""
        hash_value = int(hashlib.md5(partition_key.encode()).hexdigest(), 16)
        return hash_value % self.num_partitions
    
    async def publish(self, event: DomainEvent, partition_key: str):
        """Publish to partition based on key."""
        partition = self._get_partition(partition_key)
        queue_name = f"events.partition.{partition}"
        
        await self.broker.publish(
            queue_name,
            Message(
                type=event.event_type,
                payload={
                    "event_id": event.event_id,
                    "aggregate_id": event.aggregate_id,
                    "data": event.data,
                },
                metadata={"partition_key": partition_key},
            ),
        )


# Usage: Events for same order go to same partition
publisher = PartitionedEventPublisher(broker)

# All events for order-123 will be in same partition (ordered)
await publisher.publish(order_created_event, partition_key="order-123")
await publisher.publish(payment_processed_event, partition_key="order-123")
await publisher.publish(order_shipped_event, partition_key="order-123")

Practical Exercise

Exercise: Build an Event-Driven Order System

# order_service.py
from dataclasses import dataclass
from enum import Enum
import asyncio


class OrderStatus(Enum):
    CREATED = "created"
    CONFIRMED = "confirmed"
    FAILED = "failed"


@dataclass
class Order:
    id: str
    customer_id: str
    items: list[dict]
    total: float
    status: OrderStatus = OrderStatus.CREATED


class OrderService:
    """Order service with event-driven architecture."""
    
    def __init__(
        self, 
        repository, 
        event_publisher: EventPublisher,
        idempotency_store: IdempotencyStore,
    ):
        self.repository = repository
        self.event_publisher = event_publisher
        self.idempotency_store = idempotency_store
    
    async def create_order(
        self, 
        customer_id: str, 
        items: list[dict],
        idempotency_key: str,
    ) -> Order:
        """Create order and publish event."""
        # Check idempotency
        if await self.idempotency_store.is_processed(idempotency_key):
            existing = await self.idempotency_store.get_result(idempotency_key)
            return Order(**json.loads(existing))
        
        # Create order
        order = Order(
            id=str(uuid.uuid4()),
            customer_id=customer_id,
            items=items,
            total=sum(item["price"] * item["quantity"] for item in items),
        )
        
        # Save order
        await self.repository.save(order)
        
        # Publish event
        await self.event_publisher.publish(
            DomainEvent.create(
                event_type="order.created",
                aggregate_id=order.id,
                aggregate_type="Order",
                data={
                    "customer_id": customer_id,
                    "items": items,
                    "total": order.total,
                },
            )
        )
        
        # Mark as processed
        await self.idempotency_store.mark_processed(
            idempotency_key,
            json.dumps(order.__dict__),
        )
        
        return order


# inventory_service.py - Event consumer
class InventoryEventHandler:
    """Handle inventory-related events."""
    
    def __init__(self, inventory_repo, event_publisher):
        self.inventory_repo = inventory_repo
        self.event_publisher = event_publisher
    
    async def handle_order_created(self, event: dict):
        """Reserve inventory for new order."""
        order_id = event["aggregate_id"]
        items = event["data"]["items"]
        
        try:
            for item in items:
                await self.inventory_repo.reserve(
                    product_id=item["product_id"],
                    quantity=item["quantity"],
                    order_id=order_id,
                )
            
            # Publish success event
            await self.event_publisher.publish(
                DomainEvent.create(
                    event_type="inventory.reserved",
                    aggregate_id=order_id,
                    aggregate_type="Order",
                    data={"items": items},
                )
            )
        
        except InsufficientInventoryError as e:
            # Publish failure event
            await self.event_publisher.publish(
                DomainEvent.create(
                    event_type="inventory.reservation_failed",
                    aggregate_id=order_id,
                    aggregate_type="Order",
                    data={"reason": str(e)},
                )
            )

Key Takeaways

1. Async for decoupling - Use when you don't need immediate responses

2. Events for integration - Publish domain events for cross-service communication

3. Idempotency is essential - Messages may be delivered multiple times

4. Eventual consistency - Design for delays and out-of-order processing

5. Monitor queues - Watch for backlogs and dead letters

What's Next?

With both sync and async patterns covered, we need a unified entry point. In Article 6: API Gateway Pattern, we'll explore how to route, authenticate, and rate limit requests at the edge.

---

This article is part of the Microservice Architecture 101 series.