RabbitMQ系列（三）基本概念之Consumer

在 RabbitMQ 中，Consumer（消费者） 是负责从队列（Queue）中获取并处理消息的客户端角色，其核心机制与功能如下：

一、Consumer 的定义与核心作用

1. 消息处理终端
   Consumer 通过订阅或拉取队列中的消息，进行业务逻辑（如数据处理、通知发送等）处理，是消息传递的最终使用者。
2. 解耦生产者与消费速度
   生产者（Publisher）只需关注消息发送，无需感知消费者的数量和处理能力，消费者(Consumer)独立按需处理消息，不直接与生产者关联，解耦他们的关系。

二、Consumer 的工作模式

1. Push 模式（订阅模式）

• 机制：通过 basicConsume 方法向队列注册订阅，RabbitMQ 主动推送消息到消费者。
• 特点：
  • 实时性高，消息到达队列后立即推送。
  • 需配合手动确认（Manual Acknowledgement）防止消息丢失1。
• 代码示例：

  channel.basicConsume(queueName,  false, "myConsumerTag", new DefaultConsumer(channel) {
      @Override 
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
          // 处理消息逻辑 
          channel.basicAck(envelope.getDeliveryTag(),  false); // 手动确认 
      }
  });

2. Pull 模式（轮询模式）

• 机制：通过 basicGet 方法主动从队列拉取消息。
• 特点：
  • 适用于低频或批量处理场景。
  • 每次调用仅获取一条消息，需循环处理。
• 代码示例：

  GetResponse response = channel.basicGet(queueName,  false);
  if (response != null) {
      // 处理消息 
      channel.basicAck(response.getEnvelope().getDeliveryTag(),  false);
  }

三、消息确认机制（Acknowledgement）

1. 自动确认（Auto-Ack）
   • 参数 autoAck=true，消息一经推送即从队列删除。
   • 风险：若消费者处理失败，消息将永久丢失13。
2. 手动确认（Manual-Ack）
   • 参数 autoAck=false，需调用 basicAck 显式确认。
   • 优势：确保消息处理成功后再删除，支持重试机制。
   • 方法：

     channel.basicAck(deliveryTag,  multiple); // 确认单条或批量消息 
     channel.basicReject(deliveryTag,  requeue); // 拒绝并重新入队（可选）

四、预取策略（Prefetch）

通过 basicQos 控制消费者同时处理的消息数，优化资源分配：

• 作用：防止单个消费者因处理速度慢导致消息堆积，提升整体的吞吐能力。
• 参数：
  • prefetchCount：允许未确认的最大消息数（如设置为 10，则最多同时处理 10 条消息）。
  • prefetchSize：消息总大小限制（通常设为 0 表示不限制）。
• 示例配置：

  channel.basicQos(10); // 每次预取 10 条消息

五、典型应用场景

1. 异步任务处理
   例如订单系统将支付成功消息推送到队列，消费者异步更新库存和发送通知。
2. 负载均衡
   多个消费者订阅同一队列，RabbitMQ 通过轮询策略平均分配消息3。
3. RPC 调用
   消费者处理请求后，通过回调队列返回结果，实现远程过程调用3。

六、注意事项

• 消费者标签（Consumer Tag）：唯一标识消费者，用于取消订阅或管理特定消费者。
• 独占队列（Exclusive Queue）：设置 exclusive=true 时，队列仅允许一个消费者连接。
• 消费者取消：通过 basicCancel 方法终止指定消费者的消息接收