1
# 直接调用：同步、耦合、无容错
2
def create_order(order):
3
    inventory_result = inventory_client.deduct(order)     # 库存服务挂了→异常
4
    notification_result = notification_client.send(order)  # 通知服务慢→阻塞
5
    points_result = points_client.add(order)              # 积分服务超时→失败
6
    return {"status": "created"}
7

8
# 消息队列：异步、解耦、容错
9
def create_order(order):
10
    # 只需发消息，不管下游状态
11
    mq.publish("order-created", order)
12
    return {"status": "created"}  # 50ms 返回
13

14
# 各下游独立消费
15
@mq.subscribe("order-created")
16
def handle_inventory(order):
17
    inventory_client.deduct(order)  # 挂了？消息重试，不丢
18

19
@mq.subscribe("order-created")
20
def handle_notification(order):
21
    notification_client.send(order)  # 慢？不影响其他消费者

1
生产者 → 队列 → 消费者（一条消息只被一个消费者处理）
2

3
适用场景：任务分发、工作队列
4
示例系统：RabbitMQ 的 Queue 模型

1
// RabbitMQ 工作队列：3 个 worker 竞争消费
2
ConnectionFactory factory = new ConnectionFactory();
3
factory.setHost("localhost");
4
Connection connection = factory.newConnection();
5
Channel channel = connection.createChannel();
6

7
// 声明队列
8
channel.queueDeclare("task-queue", true, false, false, null);
9

10
// 发布任务
11
channel.basicPublish("", "task-queue",
12
    MessageProperties.PERSISTENT_TEXT_PLAIN,
13
    "process-order-12345".getBytes());
14

15
// 消费任务（3 个 worker 各自运行此代码）
16
channel.basicQos(1); // 公平分发：一次只拿一条
17
channel.basicConsume("task-queue", false, (tag, delivery) -> {
18
    String task = new String(delivery.getBody());
19
    processTask(task);
20
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
21
}, tag -> {});

1
生产者 → Topic → 多个消费者组（每条消息被每个组处理一次）
2

3
适用场景：事件通知、数据分发
4
示例系统：Kafka 的 Topic + Consumer Group 模型

1
# Kafka 发布/订阅：订单事件被多个组独立消费
2
from kafka import KafkaProducer, KafkaConsumer
3

4
# 生产者：发布订单事件
5
producer = KafkaProducer(bootstrap_servers='localhost:9092')
6
producer.send('order-events', key=b'order-123', value=b'created')
7

8
# 消费者组1：库存服务
9
inventory_consumer = KafkaConsumer(
10
    'order-events',
11
    group_id='inventory-service',
12
    bootstrap_servers='localhost:9092'
13
)
14

15
# 消费者组2：通知服务（独立消费，不影响库存组）
16
notification_consumer = KafkaConsumer(
17
    'order-events',
18
    group_id='notification-service',
19
    bootstrap_servers='localhost:9092'
20
)

1
// Kafka 事件流：消费者可以重放
2
reader := kafka.NewReader(kafka.ReaderConfig{
3
    Brokers:   []string{"localhost:9092"},
4
    Topic:     "order-events",
5
    GroupID:   "analytics-service",
6
    // 从最早的消息开始消费（重放）
7
    StartOffset: kafka.FirstOffset,
8
})

1
# 同一系统中混合使用三种模型
2
# 1. 发布/订阅：订单事件广播
3
kafka_producer.send('order-events', order_event)
4

5
# 2. 点对点：图片处理任务分发
6
rabbitmq_channel.basic_publish('', 'image-processing', image_task)
7

8
# 3. 事件流：用户行为日志持久化
9
kafka_producer.send('user-behaviors', behavior_event)

1
Topic: orders (3 partitions)
2

3
Partition 0: [msg0] [msg3] [msg6] [msg9]  ← offset 递增
4
Partition 1: [msg1] [msg4] [msg7] [msg10]
5
Partition 2: [msg2] [msg5] [msg8] [msg11]
6

7
同一个 key 的消息总是路由到同一个 Partition
8
key="order-123" → hash("order-123") % 3 = Partition 1

1
// 日志型（Kafka）：消费者主动拉取，控制 offset
2
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
3
for (ConsumerRecord<String, String> record : records) {
4
    process(record);
5
}
6
consumer.commitSync(); // 主动提交 offset
7

8
// 队列型（RabbitMQ）：服务端推送，ACK 后删除
9
channel.basicConsume("task-queue", false, (tag, delivery) -> {
10
    process(new String(delivery.getBody()));
11
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // ACK 后删除
12
}, tag -> {});

1
# 1. 共享数据库：最简单，最耦合
2
def create_order(order):
3
    db.execute("INSERT INTO orders VALUES (?)", order)
4
    # 库存服务轮询 orders 表，或者直接读同一数据库
5
    # 问题：库存服务直接依赖订单表结构
6

7
# 2. 文件传输：解耦，但不实时
8
def export_orders():
9
    with open("/shared/orders_20261002.csv", "w") as f:
10
        f.write(serialize(orders))
11
    # 库存服务定时扫描 /shared/ 目录
12
    # 问题：延迟高，文件格式变更需要协调
13

14
# 3. API 调用：实时，但耦合
15
def create_order(order):
16
    db.execute("INSERT INTO orders VALUES (?)", order)
17
    inventory_api.deduct(order)   # 同步调用，库存服务挂了就失败
18
    notification_api.send(order)  # 同步调用，通知服务慢就阻塞
19

20
# 4. 消息队列：解耦+异步+削峰
21
def create_order(order):
22
    db.execute("INSERT INTO orders VALUES (?)", order)
23
    mq.publish("order-created", order)  # 异步发送，立即返回
24
    # 各下游独立消费，互不影响
25

26
# 5. 事件流：可重放+审计+溯源
27
def create_order(order):
28
    event = {"type": "OrderCreated", "data": order, "timestamp": now()}
29
    kafka_producer.send("order-events", key=order.id, value=event)
30
    # 事件持久化，可重放，可审计

1
// RabbitMQ 反压：限制预取数量
2
channel.basicQos(10);  // 一次最多预取 10 条消息
3
// 消费者处理完一条，RabbitMQ 才会推送下一条
4
// 如果消费者处理慢，RabbitMQ 不会继续推送
5

6
// Kafka 反压：消费者自己控制拉取频率
7
while (true) {
8
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
9
    for (ConsumerRecord<String, String> record : records) {
10
        process(record);  // 处理慢 → 下次 poll 间隔长 → 自然反压
11
    }
12
    consumer.commitSync();
13
}