乐趣区

rabbitmq之ACK消息确认机制

jiezi
  1. ACK 消息确认机制

1.1 消费者确认
消费者确认，又可以叫消费者应答，它指的是 RabbitMQ 需要确认消息到底有没有被收到
1.1.1 自动应答
boolean autoAck = true;
channel.basicConsume(QUEUE, autoAck, defaultConsumer);

在订阅消息的时候可以指定应答模式, 当自动应答等于 true 的时候，表示当消费者一收到消息，消息就是已经被消费了，消费者收到了消息就会立即从队列中删除。

生产者：

[Java] 纯文本查看 复制代码
?

public class Producer1 {

// 队列名称
private static final String QUEUE="helloWord";
public static void main(String[] args) throws IOException,TimeoutException {
    Connection connection=null;
    Channel channel=null;

    try {
        // 创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //rabbitmq 服务所在 ip
        connectionFactory.setHost("192.168.56.130");
        //rabbitmq 的连接端口
        connectionFactory.setPort(5672); 
        connectionFactory.setUsername("guest");// 账户名
        connectionFactory.setPassword("guest");// 密码
        // 当前账户绑定的虚拟主机
        connectionFactory.setVirtualHost("/");
        // 获取连接
        connection = connectionFactory.newConnection();
        // 创建与交换机的通道, 可以创建多个, 每一个通道代表一个会话任务
        channel  = connection.createChannel();

// 绑定队列

        channel.queueDeclare(QUEUE,true,false,false,null);
        String message="hello:"+System.currentTimeMillis();

// 发送消息

        channel.basicPublish("",QUEUE,null,message.getBytes());
        System.out.println("send+++++++++++++++++");
    } catch (Exception e) {e.printStackTrace();
    } finally {channel.close();
        connection.close();}
}

}

消费者：
[Java] 纯文本查看 复制代码
?

public class consumer1 {

// 队列名称
private static final String QUEUE="helloWord";

public static void main(String[] args) throws IOException,TimeoutException {
    // 创建连接工程
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("192.168.56.130");
    connectionFactory.setPort(5672);
    connectionFactory.setUsername("guest");
    connectionFactory.setPassword("guest");
    connectionFactory.setVirtualHost("/");
    Connection connection = connectionFactory.newConnection();
    final Channel channel = connection.createChannel();
    // 声明队列
    channel.queueDeclare(QUEUE, true, false, false, null);

    // 定义消费方法
    DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag,Envelope envelope,
          AMQP.BasicProperties properties,byte[] body)throws IOException

{

         // 交换机
         String exchange = envelope.getExchange();
         // 路由 key
         String routingKey = envelope.getRoutingKey();
         // 消息 id
         long deliveryTag = envelope.getDeliveryTag();
         // 消息内容
         String msg = new String(body, "utf-8");
         System.out.println("receive message.." + msg);
        }
    };

    /**
     * 监听队列：*  1. 队列名称
     *  2. 是否自动回复，设置为 true 自动进行回复，mq 接收到回复会删除消息
     *                  设置为 false 需要手动进行确认消息
     *  3. 消费消息的方法，消费者接收到消息后调用此方法，获取消息
     */
    boolean autoAck = true;

channel.basicConsume(QUEUE, autoAck, defaultConsumer);

}

}

运行生产者：
我们运行生产者代码，运行 5 次，我们可以看到 Ready=5, Unacked=0, Total=5, Total 代表队列中的消息总条数，Ready 代表消费者还可以读到的条数，Unacked: 代表还有多少条没有被应答。

运行消费者：
在消费者端的获取消息的第一行打个断点，可以看到，第一次进入到 handleDelivery()方法时，队列瞬间被清空。Ready=0, Unacked=0, Total=0

Rabbitmq 控制台：

当消费者连接上队列了，因为没有指定消费者一次获取消息的条数，所以队列把队列中的所有消息一下子推送到消费者端，当消费者订阅的该队列，消息就会从队列推到客户端，当消息从队列被推出的时的那一刻就表示已经对消息进行自动确认了，消息就会从队列中删除。

服务器宕机消息丢失：
继续运行程序成功消费了一条消息，程序重新进入 handleDelivery()方法消费第二条消息，这个时候服务器突然宕机，那么你就会丢失余下的所有消息。

模拟服务器宕机：
打开任务管理器，点击详细信息，找到 java.exe 的进程，结束它。

回到我们编辑器，发现程序已经终止，但是消息只消费了一条，查看 rabbitmq 的管理界面，发现消息已经被删除，我们丢失了 4 条消息，这种情况将会产生严重的生产事故。
那我们如何去解决这种问题的，这时候我们就需要使用到手动应答。

1.1.2 手动应答
[Java] 纯文本查看 复制代码
?

boolean autoAck = false;
channel.basicConsume(QUEUE, autoAck, defaultConsumer);

手动应答和自动应答不一样，需要将 autoAck 设置为 false，当消费者收到消息在合适的时候来显示的进行确认，说我已经接收到了该消息了，RabbitMQ 可以从队列中删除该消息了。

可以调用 channel.basicAck(envelope.getDeliveryTag(), false); 告诉 rabbitmq 删除消息，该方法添加在消息消费后，如下代码：

public class consumer1 {
// 队列名称
private static final String QUEUE=”helloWord”;
public static void main(String[] args) throws IOException, TimeoutException {

 // 创建连接工程
 ConnectionFactory connectionFactory = new ConnectionFactory();
 connectionFactory.setHost("192.168.56.130");
 connectionFactory.setPort(5672);
 connectionFactory.setUsername("guest");
 connectionFactory.setPassword("guest");
 connectionFactory.setVirtualHost("/");
 Connection connection = connectionFactory.newConnection();
 final Channel channel = connection.createChannel();
  // 声明队列
 channel.queueDeclare(QUEUE, true, false, false, null);
 // 定义消费方法
 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
           // 交换机
           String exchange = envelope.getExchange();
           // 路由 key
           String routingKey = envelope.getRoutingKey();
           // 消息 id
           long deliveryTag = envelope.getDeliveryTag();
           // 消息内容
           String msg = new String(body, "utf-8");
           System.out.println("receive message.." + msg);

  // 消费者在消费消息后进行手动确认
           channel.basicAck(envelope.getDeliveryTag(), false);
        }
    };
    boolean autoAck = false;
    channel.basicConsume(QUEUE, autoAck, defaultConsumer);
}

}
测试效果：
运行 5 次生产者生产 5 条消息，在消费者端的获取消息的第一行打个断点，代码执行完 channel.basicConsume(QUEUE_NAME, false, consumer); 还没有进入到 handleDelivery()方法时可以看到 Ready=0, Unacked=5, Total=5

当代码进入 handleDelivery()方法消费了一条消息后，每执行一次 channel.basicAck(envelope.getDeliveryTag(), false)，Unacked 和 Total 就会减去 1，直到两个值都为 0

Rabbitmq 管理控制台：

服务器宕机消息还原：
这时候如果服务器单机消息为被消费的消息将会还原，结束 java.exe 进程，然后查看 rabbitmq 控制台。服务器重启消息继续正常消费。

退出移动版