消息确认机制---confirm异步

一：介绍

1.异步模式介绍

　　Channel对象提供ConfirmListener()回调方法只包含deliverTag（当前Channel发出的序列号），我们需要自己为每一个Channel维护一个unconfirm的消息序列集合，没publish一条数据，集合就加1，每回调一次handleAck方法，unconfirm集合删掉相应的一条（multiple=false）或者多条（multiple=true）记录。从程序运行效率上看，这个unconfirm集合最好采用有序集合SortedSet存储结构。

二：程序

1.生产者

 package com.mq.AsynConfirm;

 import com.mq.utils.ConnectionUtil;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.ConfirmListener;

 import com.rabbitmq.client.Connection;

 import java.io.IOException;

 import java.util.Collections;

 import java.util.SortedSet;

 import java.util.TreeSet;

 public class Send {

     private static final String QUEUE_NAME="test_queue_confirm_asyn";

     public static void main(String[] args)throws Exception{

         Connection connection= ConnectionUtil.getConnection();

         Channel channel=connection.createChannel();

         channel.queueDeclare(QUEUE_NAME,false,false,false,null);

         //生产者调用confirmSelect将channel设置为nconfirm模式

         channel.confirmSelect();

         final SortedSet<Long> confirmSet= Collections.synchronizedSortedSet(new TreeSet<Long>());

         channel.addConfirmListener(new ConfirmListener() {

             //没有问题

             public void handleAck(long deliveryTag, boolean multiple) throws IOException {

                 if (multiple){

                     System.out.println("handleAck multiple");

                     confirmSet.headSet(deliveryTag+1).clear();

                 }else{

                     System.out.println("handleAck false");

                     confirmSet.remove(deliveryTag);

                 }

             }

             //有问题

             public void handleNack(long deliveryTag, boolean multiple) throws IOException {

                 if (multiple){

                     System.out.println("handleNack multiple");

                     confirmSet.headSet(deliveryTag+1).clear();

                 }else{

                     System.out.println("handleNack false");

                     confirmSet.remove(deliveryTag);

                 }

             }

         });

         String msg="success";

         while (true){

             long seqNo=channel.getNextPublishSeqNo();

             channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());

             confirmSet.add(seqNo);

         }

     }

 }

2.消费者

 package com.mq.AsynConfirm;

 import com.mq.utils.ConnectionUtil;

 import com.rabbitmq.client.*;

 import java.io.IOException;

 public class Receive {

     private static final String QUEUE_NAME="test_queue_confirm_asyn";

     public static void main(String[] args)throws Exception {

         Connection connection = ConnectionUtil.getConnection();

         Channel channel = connection.createChannel();

         channel.queueDeclare(QUEUE_NAME, false, false, false, null);

         channel.basicConsume(QUEUE_NAME,true,new DefaultConsumer(channel){

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                 System.out.println(new String(body,"utf-8"));

             }

         });

     }

 }

3.现象

　　Send：

巴特西

消息确认机制---confirm异步

最新文章

热门文章