生产者端confirm模式的实现原理

生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布
的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，
broker就会发送一个确认给生产者 (包含消息的唯一ID) ,这就使得生产者知道消息已经正
确到达目的队列了,如果消息和队列是可持久化的，那么确认消息会将消息写入磁盘之后发出，broker回传给生产者的确认消息中deliver-tag 域包含了确认消息的序列号，此外
broker也可以设置basic.ack的multiple 域，表示到这个序列号之前的所有消息都已经得
到了处理。。

Confrm模式最大的好处在于他是异步

开启confirm模式

channel.confimSelect()

编程模式:

1.普通： 发一条waitForConfirms()
2.批量的：发一批waitForConfirms()
3.异步: confirm模式:提供一个回调。

confirm单条

• 生产者

package com.blj.rabbitmq.confirm;

import com.blj.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/** * 生产者 * confirm 普通模式 * * @author BaiLiJun on 2019/10/28 0028 */
public class Send1 { 

    private static final String QUEUE_NAME = "test_queue_confirm1";

    public static void main(String[] args) throws Exception { 
        Connection connetion = ConnectionUtils.getConnetion();
        Channel channel = connetion.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //生产者调用confirmselect将channel 设置为conf1xm模式
        //注意confirm模式跟事务机制不能在同一个队列中
        channel.confirmSelect();
        String msg = "hello confirm message!";
        channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
        if (!channel.waitForConfirms()) { 
            System.out.println("confirm message send failed");
        } else { 
            System.out.println("confirm message send ok");
        }
        channel.close();
        connetion.close();
    }

}

• 消费者

package com.blj.rabbitmq.confirm;

import com.blj.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/** * 消费者1 * confirm 普通模式 * * @author BaiLiJun on 2019/10/27 0027 */
public class Recv1 { 
    private static final String QUEUE_NAME = "test_queue_confirm1";

    public static void main(String[] args) throws IOException, TimeoutException { 
        Connection connetion = ConnectionUtils.getConnetion();
        Channel channel = connetion.createChannel();
        //队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //声明消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) { 
            //消息到达，触发此方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                String msg = new String(body, "utf-8");
                System.out.println("[confirm] Recv msg = " + msg);
            }
        };

        //监听队列
        channel.basicConsume(QUEUE_NAME, true, consumer);

    }

}

confirm批量

• 生产者

package com.blj.rabbitmq.confirm;

import com.blj.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/** * 生产者 * confirm 批量模式 * * @author BaiLiJun on 2019/10/28 0028 */
public class Send2 { 

    private static final String QUEUE_NAME = "test_queue_confirm2";

    public static void main(String[] args) throws Exception { 
        Connection connetion = ConnectionUtils.getConnetion();
        Channel channel = connetion.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //生产者调用confirmselect将channel 设置为conf1xm模式
        //注意confirm模式跟事务机制不能在同一个队列中
        channel.confirmSelect();
        String msg = "hello confirm message!";
        //批量发送
        for (int i = 0; i <10 ; i++) { 
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
        }
        //确认
        if (!channel.waitForConfirms()) { 
            System.out.println("confirm message send failed");
        } else { 
            System.out.println("confirm message send ok");
        }
        channel.close();
        connetion.close();
    }

}

• 消费者

package com.blj.rabbitmq.confirm;

import com.blj.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/** * 消费者1 * confirm 普通模式 * * @author BaiLiJun on 2019/10/27 0027 */
public class Recv1 { 
    private static final String QUEUE_NAME = "test_queue_confirm2";

    public static void main(String[] args) throws IOException, TimeoutException { 
        Connection connetion = ConnectionUtils.getConnetion();
        Channel channel = connetion.createChannel();
        //队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //声明消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) { 
            //消息到达，触发此方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                String msg = new String(body, "utf-8");
                System.out.println("[confirm] Recv msg = " + msg);
            }
        };

        //监听队列
        channel.basicConsume(QUEUE_NAME, true, consumer);

    }

}

异步模式

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

• 生产者

package com.blj.rabbitmq.confirm;

import com.blj.rabbitmq.util.ConnectionUtils;
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;

/** * 生产者 * confirm 异步模式 * * @author BaiLiJun on 2019/10/28 0028 */
public class Send3 { 

    private static final String QUEUE_NAME = "test_queue_confirm3";

    public static void main(String[] args) throws Exception { 
        Connection connetion = ConnectionUtils.getConnetion();
        Channel channel = connetion.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //生产者调用confirmselect将channel 设置为conf1xm模式
        //注意confirm模式跟事务机制不能在同一个队列中
        channel.confirmSelect();

        //未确认的消息标识
        SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());

        //通过添加监听
        channel.addConfirmListener(new ConfirmListener() { 
           //没有问题的handleAck
            @Override
            public void handleAck(long deliveryTag, boolean multip1e) throws IOException { 
                if(multip1e){ 
                    System.out.println("----handleAck----multip1e");
                    confirmSet.headSet(deliveryTag+1).clear();
                }else { 
                    System.out.println("----handleAck----multip1e false");
                    confirmSet.remove(deliveryTag);
                }
            }

           //handleNack
            @Override
            public void handleNack(long deliveryTag, boolean multip1e) throws IOException { 
                if(multip1e){ 
                    System.out.println("----handleNack----multip1e");
                    confirmSet.headSet(deliveryTag+1).clear();
                }else { 
                    System.out.println("----handleNack----multip1e false");
                    confirmSet.remove(deliveryTag);
                }
            }
        });

        String msg = "hello confirm message!";
        while (true){ 
            long seqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
            confirmSet.add(seqNo);
        }

    }

}

• 消费者

package com.blj.rabbitmq.confirm;

import com.blj.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/** * 消费者3 * confirm 异步模式 * * @author BaiLiJun on 2019/10/27 0027 */
public class Recv3 { 
    private static final String QUEUE_NAME = "test_queue_confirm3";

    public static void main(String[] args) throws IOException, TimeoutException { 
        Connection connetion = ConnectionUtils.getConnetion();
        Channel channel = connetion.createChannel();
        //队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //声明消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) { 
            //消息到达，触发此方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                String msg = new String(body, "utf-8");
                System.out.println("[confirm] Recv msg = " + msg);
            }
        };

        //监听队列
        channel.basicConsume(QUEUE_NAME, true, consumer);

    }

}