目录

 

 

基本概念

代码与实例

---

 

基本概念

异步模式：Channel对象提供的ConfirmListener()回调方法只包含deliverTag(当前Channel发出的消息序列号)，需要自己为每一个Channel维护一个cunconfirm的消息序列号集合，每个publish数据，集合中元素+1，回调一次handleAck方法，unconfirm集合删除相应的一条（multiple=false）或多条（multiple=true）记录。从程序效率上看，这个unconfirm集合最好采用有序集合SortedSet存储结构

 

代码与实例

程序运行截图如下：

源码如下：

package tx;import com.rabbitmq.client.Channel; import com.rabbitmq.client.ConfirmListener; import com.rabbitmq.client.Connection; import util.ConnectionUtils;import java.io.IOException; import java.util.Collections; import java.util.SortedSet; import java.util.TreeSet; import java.util.concurrent.TimeoutException;public class TxSend {private static final String QUEUE_NAME = "test_queue_confirm3";public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Connection connection = ConnectionUtils.getConnect();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);//生产者调用confirmSelect 将 channel 设置为 confirm 模式channel.confirmSelect();//未确认的标识final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());channel.addConfirmListener(new ConfirmListener() {//ACKpublic void handleAck(long l, boolean b) throws IOException {if(b){System.out.println("> handleAck multiple!"); //多条confirmSet.headSet(l + 1).clear();}else{System.out.println("> handleAck multiple false!"); //单条confirmSet.remove(l);}}//Nack 回执有问题的public void handleNack(long l, boolean b) throws IOException {if(b){System.out.println("> handleNack multiple!"); //多条confirmSet.headSet(l + 1).clear();}else{System.out.println("> handleNack multiple false!"); //单条confirmSet.remove(l);}}});String msgStr = "Hello World!";while (true){long seqNo = channel.getNextPublishSeqNo();channel.basicPublish("", QUEUE_NAME, null, msgStr.getBytes());confirmSet.add(seqNo);}} }

 

 

总结

以上是生活随笔为你收集整理的Java笔记-RabbitMQ中生产者端confirm模式(异步模式)的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。