Rabbitmq消息确认机制

1.生产者确认机制

确认消息发送到交换机--Confirm方式

1.1普通Confirm方式

private static void sendMsg(Channel channel) throws IOException, InterruptedException {
        //开启确认机制
        channel.confirmSelect();
        //发送消息到exchange
        String msg = "hello confirm";
 
        channel.basicPublish("", "no-lost", MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
 
        if (channel.waitForConfirms()) {
            System.out.println("生产者发布消息至Exchange成功！");
        } else {
            System.out.println("生产者发布消息至Exchange失败！请重试");
        }
    }
 

1.2异步Confirm方式

 private static void sendMsg(Channel channel) throws IOException, InterruptedException {
        //开启确认机制
        channel.confirmSelect();
        //发送消息到exchange
        String msg = "hello confirm";
        for (int i = 0; i < 10; i++) {
            channel.basicPublish("", "no-lost", MessageProperties.PERSISTENT_TEXT_PLAIN, (msg + i).getBytes());
        }
 
        //开启异步回调
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("生产者发布消息至Exchange成功，标示为：" + deliveryTag + ",是否为批量操作：" + multiple);
            }
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("生产者发布消息至Exchange失败，标示为：" + deliveryTag + ",是否为批量操作：" + multiple);
            }
        });
        System.in.read();
    }

2.交换机到消息队列的return确认机制

 //3开启Return机制
        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //当送达失败是才会回调
                System.out.println(new String(body, "utf-8") + ",消息没有送达到queue中");
            }
        });
 

3.消费者确认机制

接收消息成功时

channel.basicAck(envelope.getDeliveryTag(), false);

接收消息失败或进入异常时

 try {
                    //具体业务
                    int i = 1 / 0;
                    //确认
                    channel.basicAck(envelope.getDeliveryTag(), false);
                } catch (Exception e) {
                    if (errorMap.get(new String(body, "UTF-8")) != null) {
                        System.out.println("消息已重复处理失败,拒绝再次接收...");
                        channel.basicReject(envelope.getDeliveryTag(), false);
                    } else {
                        System.out.println("消息即将再次返回队列处理...");
                        channel.basicNack(envelope.getDeliveryTag(), false, true);
                        errorMap.put(new String(body, "UTF-8"), 1);
                    }
                }
            }
 

一般在该消息处理完后执行，该消息才会在队列里面被删除，不然会处于UnAcked的状态存在队列中

finally {channel.basicAck(envelope.getDeliveryTag(), false);
}