Understanding Event-Driven Architecture: Core Concepts, Message Queues, and Design Patterns

1. Event‑Driven Overview

1.1 What Is Event‑Driven

In event‑driven architecture, systems communicate via publish/subscribe events, which decouples producers from consumers.

传统请求驱动 vs 事件驱动
┌─────────────────────────────────────────────────────────────┐
│                                                         │
│  请求驱动：                                            │
│  ┌─────────┐          直接调用  ┌─────────┐            │
│  │ 订单   │ ──────►   调用   │ 库存   │            │
│  │ 服务   │                │ 服务   │            │
│  └─────────┘                └─────────┘            │
│                                                         │
│                     ▼                                   │
│               ┌─────────┐                              │
│               │ 物流    │                              │
│               │ 服务    │                              │
│               └─────────┘                              │
│                                                         │
│  事件驱动：                                            │
│  ┌─────────┐          发布事件  ┌─────────────────────┐ │
│  │ 订单   │ ──► 发布事件 ──► │   消息代理/总线      │ │
│  │ 服务   │                │                     │ │
│  └─────────┘                └─────────────────────┘ │
│                               │   │               │
│                               ▼   ▼               ▼
│                         ┌─────────┐ ┌─────────┐ ┌─────────┐
│                         │ 库存    │ │ 物流    │ │ 通知    │
│                         │ 消费者 │ │ 消费者 │ │ 消费者 │
│                         └─────────┘ └─────────┘ └─────────┘
│                                                         │
└─────────────────────────────────────────────────────────────┘

1.2 Advantages of Event‑Driven

Decoupling : Producers and consumers evolve independently.

Extensibility : New consumers can be added easily.

Fault‑tolerance : Message persistence guarantees delivery.

Performance : Asynchronous processing smooths traffic spikes.

Traceability : Event sourcing enables end‑to‑end tracing.

1.3 Core Concepts

Event : An immutable fact.

Producer : Publishes events.

Consumer : Subscribes to events.

Topic/Channel : Event channel.

Broker : Message middleware.

2. Message Queues

2.1 AWS SQS

import boto3
import json

sqs = boto3.client('sqs')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789/my-queue'

# 发送消息
def send_order_event(order_id, event_type):
    response = sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps({
            'orderId': order_id,
            'eventType': event_type,
            'timestamp': '2024-01-01T10:00:00Z'
        }),
        MessageAttributes={
            'eventType': {
                'StringValue': event_type,
                'DataType': 'String'
            }
        }
    )
    return response['MessageId']

# 消费消息
def process_messages():
    while True:
        response = sqs.receive_message(
            QueueUrl=queue_url,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=20
        )
        for msg in response.get('Messages', []):
            body = json.loads(msg['Body'])
            # 处理消息
            process_order_event(body)
            # 删除消息
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=msg['ReceiptHandle']
            )

2.2 RabbitMQ

import pika
import json

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# 声明交换机和队列
channel.exchange_declare(
    exchange='orders',
    exchange_type='topic',
    durable=True
)

channel.queue_declare(queue='order_events', durable=True)

# 绑定队列到交换机
channel.queue_bind(
    exchange='orders',
    queue='order_events',
    routing_key='order.*'
)

# 发布消息
def publish_order_event(event_type, order_id, data):
    channel.basic_publish(
        exchange='orders',
        routing_key=f'order.{event_type}',
        body=json.dumps({
            'orderId': order_id,
            'eventType': event_type,
            'data': data
        }),
        properties=pika.BasicProperties(
            delivery_mode=2,  # 持久化
            content_type='application/json'
        )
    )

# 消费消息
def callback(ch, method, properties, body):
    data = json.loads(body)
    print(f"Received: {data}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='order_events', on_message_callback=callback)
channel.start_consuming()

2.3 Apache Kafka

from confluent_kafka import Producer, Consumer, KafkaError

# 生产者
producer = Producer({
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'order-service'
})

def send_order_event(order):
    producer.produce(
        'order-events',
        key=order['id'].encode(),
        value=json.dumps(order).encode()
    )
    producer.flush()

# 消费者
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'inventory-service',
    'auto.offset.reset': 'earliest'
})

consumer.subscribe(['order-events'])

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None:
        continue
    if msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        else:
            raise KafkaException(msg.error())
    order = json.loads(msg.value().decode())
    process_order(order)
    consumer.commit(msg)

3. EventBridge

3.1 EventBridge Architecture

AWS EventBridge
┌─────────────────────────────────────────────────────────────┐
│                                                         │
│  ┌─────────┐      ┌─────────────┐      ┌─────────────┐ │
│  │ AWS     │      │             │      │  Lambda      │ │
│  │ Service │ ───► │  Event      │ ───► │  Function    │ │
│  │         │      │  Bus        │      │              │ │
│  └─────────┘      │             │      └─────────────┘ │
│                    │  ┌───────┐ │      ┌─────────────┐ │
│  ┌─────────┐      │  │ Rule 1│ │ ───► │  Kinesis    │ │
│  │ Custom  │ ───► │  ├───────┤ │      │  Data Stream│ │
│  │ App     │      │  │ Rule 2│ │ ───► │             │ │
│  └─────────┘      │  ├───────┤ │      └─────────────┘ │
│                    │  │ Rule 3│ │      ┌─────────────┐ │
│                    │  └───────┘ │ ───► │  SQS Queue  │ │
│                    └─────────────┘      └─────────────┘ │
│                                                         │
└─────────────────────────────────────────────────────────────┘

3.2 EventBridge Events

import boto3
import json

eventbridge = boto3.client('events')

# 发布事件到自定义总线
def publish_order_event(order):
    eventbridge.put_events(
        Entries=[{
            'Source': 'com.myapp.orders',
            'DetailType': 'OrderCreated',
            'Detail': json.dumps(order),
            'EventBusName': 'default'
        }]
    )

# 创建规则
eventbridge.create_rule(
    Name='order-created-rule',
    EventPattern=json.dumps({
        'source': ['com.myapp.orders'],
        'detail-type': ['OrderCreated']
    }),
    Targets=[{
        'Id': 'lambda-target',
        'Arn': 'arn:aws:lambda:us-east-1:123456789:function:process-order',
        'InputTransformer': {
            'InputPathsMap': {
                'orderId': '$.detail.orderId'
            },
            'InputTemplate': "\"Order <orderId> has been created\""
        }
    }]
)

4. Event Design Patterns

4.1 Event Structure

{
  "specversion": "1.0",
  "type": "com.myapp.orders.created",
  "source": "/orders/service",
  "id": "unique-event-id",
  "time": "2024-01-01T10:00:00Z",
  "datacontenttype": "application/json",
  "data": {
    "orderId": "order-123",
    "customerId": "customer-456",
    "totalAmount": 99.99,
    "items": [
      {"productId": "prod-001", "quantity": 2}
    ]
  }
}

4.2 CloudEvents Specification

from dataclasses import dataclass
from datetime import datetime
import uuid

@dataclass
class CloudEvent:
    spec_version: str = "1.0"
    type: str = ""
    source: str = ""
    id: str = ""
    time: str = ""

    def to_dict(self):
        return {
            "specversion": self.spec_version,
            "type": self.type,
            "source": self.source,
            "id": self.id or str(uuid.uuid4()),
            "time": self.time or datetime.utcnow().isoformat()
        }

class OrderEvent(CloudEvent):
    def __init__(self, order_id, event_type, data):
        super().__init__(
            type=f"com.myapp.orders.{event_type}",
            source="/orders/service"
        )
        self.order_id = order_id
        self.data = data

4.3 Saga Pattern

# 订单 Saga 编排
class OrderSaga:
    def __init__(self, event_bus):
        self.event_bus = event_bus
        self.steps = [
            ('reserve_inventory', self.reserve_inventory),
            ('process_payment', self.process_payment),
            ('create_shipment', self.create_shipment)
        ]

    async def execute(self, order):
        saga_state = {'order': order, 'completed': [], 'failed': []}
        for step_name, step_func in self.steps:
            try:
                result = await step_func(saga_state)
                saga_state['completed'].append(step_name)
                self.event_bus.publish(f'saga.{order.id}.{step_name}.success', result)
            except Exception as e:
                saga_state['failed'].append(step_name)
                await self.compensate(saga_state)
                self.event_bus.publish(f'saga.{order.id}.failed', {'step': step_name})
                return False
        return True

    async def compensate(self, saga_state):
        for step_name in reversed(saga_state['completed']):
            compensations = {
                'reserve_inventory': self.release_inventory,
                'process_payment': self.refund_payment,
                'create_shipment': self.cancel_shipment
            }
            await compensations[step_name](saga_state)