learn-network

Appearance

AMQP 协议详解 ​

AMQP（Advanced Message Queuing Protocol）是应用层协议的标准协议，为面向消息的中间件设计。本文深入解析 AMQP 协议原理、RabbitMQ 核心概念，以及与 JMS、Kafka 的对比。

1. AMQP 协议概述 ​

1.1 什么是 AMQP ​

AMQP（高级消息队列协议）是一个用于消息中间件的对等协议，它定义了：

  • 消息格式 - 消息如何编码和封装
  • 传输协议 - 消息如何在网络上传输
  • 会话机制 - 消息的确认和事务支持
  • 安全模型 - 认证和授权机制

1.2 AMQP 历史 ​

年份版本重要事件
20040-8AMQP 规范首次发布（摩根大通）
20060-9完善元协议和扩展性
20070-9-1RabbitMQ 1.0 发布
20111.0完全重写，打破向后兼容

注意： AMQP 0-9-1 和 AMQP 1.0 是完全不兼容的协议版本。业界主流使用的是 0-9-1（也被称为 0-9，因为 1.0 改动太大未被广泛采用）。

2. AMQP 协议模型 ​

2.1 核心组件 ​

+----------------+       +----------------+       +----------------+
|   Producer     |       |   RabbitMQ     |       |   Consumer     |
|   ( Publisher )|       |   Broker       |       |   ( Subscriber )|
+--------+-------+       +-------+--------+       +-------+--------+
         |                       |                       |
         |  1. Connection         |                       |
         +---------------------->|                       |
         |                       |                       |
         |  2. Channel           |                       |
         |  (Virtual Connection)  |                       |
         +---------------------->|                       |
         |                       |                       |
         |  3. Declare Exchange  |                       |
         +---------------------->|                       |
         |                       |                       |
         |  4. Bind Queue        |                       |
         |  to Exchange          |                       |
         +---------------------->|                       |
         |                       |                       |
         |  5. Publish Message    |                       |
         +---------------------->|                       |
         |                       |                       |
         |                       |--------+              |
         |                       |        | 6. Deliver   |
         |                       |        |   Message     |
         |                       |<-------+--------------+
         |                       |                       |
         |  7. ACK (optional)    |                       |
         |<----------------------|                       |
         |                       |                       |

2.2 组件详解 ​

组件描述作用
Producer消息生产者创建并发送消息到 Exchange
Consumer消息消费者从 Queue 接收并处理消息
Broker消息中间件服务器接收、路由、存储消息
ConnectionTCP 物理连接客户端与 Broker 之间的长连接
Channel虚拟连接在 Connection 内部的多路复用通道
Exchange交换机接收消息并根据规则路由到 Queue
Queue队列存储消息的 FIFO 结构
Binding绑定关系定义 Exchange 到 Queue 的路由规则

2.3 Connection 与 Channel ​

Connection (TCP Connection)
    │
    ├── Channel 1
    ├── Channel 2
    ├── Channel 3
    └── Channel N

为什么需要 Channel：

  • TCP 连接建立成本高（需要三次握手）
  • 操作系统对单进程打开的文件描述符有限制
  • Channel 是轻量级的虚拟连接，共享同一个 TCP 连接
  • 每个 Channel 有独立的 ID，在协议层面区分

3. Exchange 交换机详解 ​

3.1 Exchange 类型 ​

AMQP 定义了四种 Exchange 类型：

类型路由规则典型用途
direct完全匹配 Routing Key点对点消息
fanout广播到所有绑定的 Queue广播消息
topic通配符匹配 Routing Key模式匹配发布
headers匹配消息头部的键值对复杂路由规则

3.2 Direct Exchange ​

精确匹配 Routing Key：

Exchange: orders.direct
    │
    ├── Binding: order.created  ──────> Queue: order-created-queue
    ├── Binding: order.paid      ──────> Queue: order-paid-queue
    └── Binding: order.cancelled ──────> Queue: order-cancelled-queue

Publish "order.created"  ──> 路由到 order-created-queue
Publish "order.paid"     ──> 路由到 order-paid-queue
Publish "order.xxx"      ──> 无匹配，消息丢弃

3.3 Fanout Exchange ​

广播到所有绑定的 Queue：

Exchange: notifications.fanout
    │
    ├── Binding ──────> Queue: email-queue
    ├── Binding ──────> Queue: sms-queue
    └── Binding ──────> Queue: push-queue

Publish message ──> 同时投递到 email-queue, sms-queue, push-queue

3.4 Topic Exchange ​

支持通配符匹配：

通配符含义示例
*精确匹配一个词stock.* 匹配 stock.usd、stock.eur
#匹配零个或多个词stock.# 匹配 stock、stock.usd、stock.usd.nyse
Exchange: logs.topic
    │
    ├── Binding: log.#        ──────> Queue: all-logs
    ├── Binding: log.error    ──────> Queue: error-logs
    └── Binding: log.*.database ────> Queue: database-logs

Publish "log.info"      ──> 匹配 log.#  ──> all-logs
Publish "log.error"     ──> 匹配 log.# 和 log.error ──> all-logs, error-logs
Publish "log.mysql.database" ──> 匹配 log.# 和 log.*.database ──> all-logs, database-logs

4. 消息确认机制 ​

4.1 Acknowledge 模式 ​

消费者确认消息有三种模式：

模式配置行为
autoautoAck=true消息投递给消费者后自动确认
manualautoAck=false消费者显式调用 basic.ack 确认
negative手动调用 basic.nack拒绝消息，可选择重新入队

4.2 手动确认流程 ​

1. Consumer                    2. Broker
    │                              │
    │  basic.deliver (msg)         │
    |<─────────────────────────────+
    │                              │
    │  处理消息...                  │
    │                              │
    │  basic.ack (delivery_tag)    │
    +────────────────────────────>-+
    │                              │
    │  (或) basic.nack             │
    │  (requeue=true)              │
    +────────────────────────────>-+  --> 消息重新入队

4.3 消息拒绝与重入队 ​

javascript
// 拒绝消息并重新入队
channel.basicNack(deliveryTag, false, true);

// 拒绝消息不重入队（死信）
channel.basicNack(deliveryTag, false, false);

// 多重拒绝
channel.basicReject(deliveryTag, true);  // true = requeue

5. 持久化与可靠性 ​

5.1 消息持久化 ​

设置选项说明
Exchange 持久化durable=trueBroker 重启后 Exchange 仍然存在
Queue 持久化durable=trueBroker 重启后 Queue 仍然存在
消息持久化deliveryMode=2消息写入磁盘
javascript
// 发布持久化消息
channel.publish('exchange.name', 'routing.key',
    Buffer.from('message content'),
    {
        persistent: true,  // 消息持久化
        contentType: 'application/json'
    }
);

5.2 publisher confirm 机制 ​

确保消息成功到达 Broker：

javascript
// 开启 publisher confirms
channel.confirmSelect();

channel.publish('exchange.name', 'routing.key',
    Buffer.from('message'),
    { persistent: true },
    function(err, ok) {
        if (err) {
            // 消息发送失败
            console.error('Message send failed:', err);
        } else {
            // Broker 已确认收到
            console.log('Message confirmed');
        }
    }
);

// 或者使用异步确认
const pendingConfirms = [];
channel.publish('exchange.name', 'routing.key',
    Buffer.from('message'),
    { persistent: true }
);
pendingConfirms.push(channel.waitForConfirms());

5.3 事务机制 ​

AMQP 支持事务（但性能开销大，不推荐在高并发场景使用）：

javascript
// 开启事务
channel.txSelect();

// 发送消息
channel.publish('exchange.name', 'routing.key', Buffer.from('msg'));

// 提交事务
channel.txCommit();
// 或回滚
channel.txRollback();

6. 消费模式 ​

6.1 推模式（Push） ​

Broker 主动推送消息给消费者：

javascript
channel.basicConsume('queue.name', false, function(msg) {
    console.log('Received:', msg.content.toString());
    channel.basicAck(msg.fields.deliveryTag, false);
});

6.2 拉模式（Pull） ​

消费者主动拉取消息：

javascript
async function consume() {
    const msg = await channel.basicGet('queue.name', false);
    if (msg) {
        console.log('Received:', msg.content.toString());
        channel.basicAck(msg.fields.deliveryTag, false);
    }
}

6.3 Prefetch 预取 ​

控制消费者未确认消息的最大数量：

javascript
// 在消费前设置 prefetch
channel.prefetch(10);  // 最多同时处理 10 条消息

channel.basicConsume('queue.name', false, function(msg) {
    // 处理消息
    channel.basicAck(msg.fields.deliveryTag, false);
});

7. 死信队列（DLX） ​

7.1 死信产生条件 ​

消息成为死信的条件：

  • 消费者拒绝且不重入队（basicReject/reject with requeue=false）
  • 消息超时（TTL 过期）
  • 队列达到最大长度

7.2 配置死信队列 ​

javascript
// 创建死信交换机
channel.assertExchange('dlx.exchange', 'direct', { durable: true });

// 创建死信队列
channel.assertQueue('dlx.queue', { durable: true });
channel.bindQueue('dlx.queue', 'dlx.exchange', 'dead.letter');

// 创建主队列，配置死信交换机
channel.assertQueue('main.queue', {
    durable: true,
    arguments: {
        'x-dead-letter-exchange': 'dlx.exchange',
        'x-dead-letter-routing-key': 'dead.letter'
    }
});

8. RabbitMQ 核心概念 ​

8.1 虚拟主机（VHost） ​

VHost 是逻辑上的隔离单元：

RabbitMQ Server
    │
    ├── VHost: /production
    │       ├── Exchange
    │       ├── Queue
    │       └── Binding
    │
    └── VHost: /development
            ├── Exchange
            ├── Queue
            └── Binding

每个 VHost 有独立的用户权限和配置。

8.2 用户与权限 ​

javascript
// 创建用户
rabbitmqctl add_user admin password;

// 设置权限
rabbitmqctl set_permissions -p / admin ".*" ".*" ".*";

// 绑定用户到 VHost
rabbitmqctl set_permissions -p /production admin ".*" ".*" ".*";

8.3 镜像队列 ​

高可用队列配置：

镜像队列副本分布：
Master Node (Primary)
    │
    ├── Slave Node 1 (Mirror)
    └── Slave Node 2 (Mirror)

所有操作在 Master 执行，同步到 Slaves

9. AMQP vs JMS vs Kafka ​

9.1 协议层面对比 ​

维度AMQPJMSKafka
协议类型底层 Wire ProtocolAPI 规范底层 Wire Protocol
跨语言原生支持需 JMS Client原生支持
模式点对点 + 发布订阅点对点 + 发布订阅仅发布订阅
消息模型Queue + ExchangeQueue + TopicPartition + Topic

9.2 架构对比 ​

AMQP/RabbitMQ 架构：

Producer ──> Exchange ──> Queue ──> Consumer
                         ↑
                    可以有多个 Consumer
                    (竞争消息)

Kafka 架构：

Producer ──> Topic ──> Partition ──> Consumer Group
                    (多副本)         (每个 Partition
                                    只能被一个 Consumer
                                    在 Group 内消费)

9.3 功能特性对比 ​

特性AMQP/RabbitMQJMSKafka
消息持久化支持取决于实现支持
事务支持支持有限支持
广播Fanout ExchangeTopic多 Consumer Group
消息优先级支持支持不支持
延迟消息插件支持取决于实现需外部实现
死信队列原生支持取决于实现需外部实现
延迟容忍中等中等高（追加写）
吞吐率中等（10K-100K/s）低高（100K-1M/s）
消息堆积受内存限制受内存限制可达 TB 级别

9.4 选型指南 ​

场景推荐原因
复杂路由规则RabbitMQ多种 Exchange 类型，通配符匹配
事务性消息RabbitMQ完整事务支持
高吞吐日志Kafka追加写，TB 级消息堆积
Java 企业应用JMS/RabbitMQ成熟生态，大量现成集成
微服务解耦RabbitMQ丰富的消息模式
流处理Kafka高吞吐，低延迟

10. 代码示例 ​

10.1 Node.js + amqplib ​

生产者代码：

javascript
const amqp = require('amqplib');

async function producer() {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createConfirmChannel();

    const exchange = 'orders.exchange';
    const routingKey = 'order.created';
    const message = {
        orderId: 'ORD-12345',
        customerId: 'CUST-001',
        amount: 99.99,
        items: [
            { productId: 'PROD-A', quantity: 2 },
            { productId: 'PROD-B', quantity: 1 }
        ]
    };

    // 声明交换机
    await channel.assertExchange(exchange, 'topic', { durable: true });

    // 发布消息（持久化）
    channel.publish(
        exchange,
        routingKey,
        Buffer.from(JSON.stringify(message)),
        {
            persistent: true,
            contentType: 'application/json',
            headers: {
                'x-source': 'order-service'
            }
        }
    );

    console.log('Message published:', message);

    await channel.close();
    await connection.close();
}

producer().catch(console.error);

消费者代码：

javascript
const amqp = require('amqplib');

async function consumer() {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createConfirmChannel();

    const queueName = 'order.created.queue';
    const exchange = 'orders.exchange';
    const routingKey = 'order.created';

    // 声明交换机
    await channel.assertExchange(exchange, 'topic', { durable: true });

    // 声明队列
    await channel.assertQueue(queueName, {
        durable: true,
        arguments: {
            'x-dead-letter-exchange': 'dlx.exchange',
            'x-dead-letter-routing-key': 'dead.letter'
        }
    });

    // 绑定队列到交换机
    await channel.bindQueue(queueName, exchange, routingKey);

    // 设置 prefetch
    await channel.prefetch(10);

    // 开始消费
    console.log('Waiting for messages...');

    channel.consume(queueName, async (msg) => {
        if (!msg) return;

        try {
            const content = JSON.parse(msg.content.toString());
            console.log('Received order:', content);

            // 模拟处理
            await processOrder(content);

            // 确认消息
            channel.basicAck(msg.fields.deliveryTag, false);

        } catch (error) {
            console.error('Processing failed:', error);
            // 拒绝消息，不重入队（进入死信队列）
            channel.basicNack(msg.fields.deliveryTag, false, false);
        }
    }, { noAck: false });
}

async function processOrder(order) {
    // 订单处理逻辑
    console.log(`Processing order ${order.orderId}...`);
    await new Promise(resolve => setTimeout(resolve, 1000));
    console.log(`Order ${order.orderId} processed successfully`);
}

consumer().catch(console.error);

10.2 Python + pika ​

消费者代码：

python
import pika
import json

credentials = pika.PlainCredentials('guest', 'guest')
parameters = pika.ConnectionParameters('localhost', 5672, '/', credentials)

connection = pika.BlockingConnection(parameters)
channel = connection.channel()

# 声明交换机和队列
channel.exchange_declare('orders.exchange', 'topic', durable=True)
channel.queue_declare('order.created.queue', durable=True)
channel.queue_bind('order.created.queue', 'orders.exchange', 'order.created')

# 设置 prefetch
channel.basic_qos(prefetch_count=10)

def callback(ch, method, properties, body):
    try:
        order = json.loads(body)
        print(f"Received order: {order}")

        # 模拟处理
        process_order(order)

        # 确认消息
        ch.basic_ack(delivery_tag=method.delivery_tag)

    except Exception as e:
        print(f"Error processing order: {e}")
        # 拒绝消息，不重入队
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

def process_order(order):
    """订单处理逻辑"""
    print(f"Processing order {order.get('orderId')}...")

# 开始消费
channel.basic_consume('order.created.queue', callback, auto_ack=False)

print('Waiting for messages...')
channel.start_consuming()

10.3 Spring Boot + RabbitMQ ​

配置类：

java
@Configuration
public class RabbitMQConfig {

    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable("order.created.queue")
                .withArgument("x-dead-letter-exchange", "dlx.exchange")
                .withArgument("x-dead-letter-routing-key", "dead.letter")
                .build();
    }

    @Bean
    public TopicExchange ordersExchange() {
        return ExchangeBuilder.topicExchange("orders.exchange")
                .durable(true)
                .build();
    }

    @Bean
    public Binding orderBinding(Queue orderQueue, TopicExchange ordersExchange) {
        return BindingBuilder.bind(orderQueue)
                .to(ordersExchange)
                .with("order.created");
    }
}

生产者：

java
@Service
public class OrderProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendOrder(Order order) {
        rabbitTemplate.convertAndSend(
            "orders.exchange",
            "order.created",
            order,
            message -> {
                message.getMessageProperties().setPersistent(true);
                message.getMessageProperties().setContentType("application/json");
                return message;
            }
        );
    }
}

消费者：

java
@Component
public class OrderConsumer {

    @RabbitListener(queues = "order.created.queue")
    public void handleOrder(Order order, Channel channel,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            processOrder(order);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("Failed to process order", e);
            channel.basicNack(deliveryTag, false, false);
        }
    }

    private void processOrder(Order order) {
        // 订单处理逻辑
    }
}

11. 常见问题与最佳实践 ​

11.1 常见问题 ​

Q: 消息丢失怎么办？ A: 确保三重持久化：

  1. 交换机和队列 durable=true
  2. 消息设置 persistent=true
  3. 开启 publisher confirm

Q: 消息重复消费怎么办？ A: 消费者做好幂等性设计：

  • 使用数据库唯一索引
  • Redis 去重
  • 消息 ID 记录

Q: 队列消息积压怎么办？ A:

  • 增加消费者数量
  • 优化消费者处理速度
  • 检查消息是否正常确认

11.2 最佳实践 ​

实践说明
使用确认机制always 开启 manual ack
合理设置 prefetch根据处理能力设置，避免积压
消息幂等处理消费者做好幂等性设计
监控队列深度队列消息数 > 1000 需告警
隔离不同环境使用不同的 VHost
连接复用避免频繁创建连接

12. 参考资源 ​