发布确认
1.设置要求队列必须持久化
2.设置要求队列中的消息
3.发布确认
什么是发布确认？
只有当消息完完整整的发送完成发布确认之后，消息才算在磁盘上保存好了，数据再怎么服务器开关都不会丢失
开启发布确认的方法
发布确认默认是没有开启的，如果要开启需要调用方法 confirmSelect，每当你要想使用发布确认，都需要在channel 上调用该方法
单个确认发布
这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布,后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。
这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。
当然对于某些应用程序来说这可能已经足够了。
//单个确认 public static void publishMessageIndividually() throws Exception{ Channel channel = RabbitMqUtils.getChannel(); //队列的声明 String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName,true,false,false,null); //开启发布确认 channel.confirmSelect(); //开始时间 long begin = System.currentTimeMillis(); //大量的发消息 for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("",queueName,null,message.getBytes()); //单个消息就马上进行发布确认 boolean flag = channel.waitForConfirms(); if(flag){ System.out.println("消息发送成功"); } } //结束时间 long end = System.currentTimeMillis(); System.out.println("发布"+MESSAGE_COUNT+"个单独确认消息，耗时"+(end-begin+"ms")); }
运行结果：
批量消息确认发布
上面我们讲到的单个确认发布消息特别慢，与其相比，先发布一批消息然后一起确认可以极大的提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。
//批量发布确认 public static void publishMessageBatch() throws Exception{ Channel channel = RabbitMqUtils.getChannel(); //队列的声明 String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName,true,false,false,null); //开启发布确认 channel.confirmSelect(); //开始时间 long begin = System.currentTimeMillis(); //批量确认消息大小 int batchSize = 100; //批量发送消息 批量发布确认 for (int i = 0; i < MESSAGE_COUNT; i++) { String message = i + ""; channel.basicPublish("",queueName,null,message.getBytes()); //判断达到100条消息的时候，批量确认一次 if(i%batchSize==0){ //发布确认 channel.waitForConfirms(); } } //结束时间 long end = System.currentTimeMillis(); System.out.println("发布"+MESSAGE_COUNT+"个批量确认消息，耗时"+(end-begin+"ms")); }
异步确认发布
异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的,这个中问件也是通过函数回调来保证是否投递成功，下面就让我们来详细讲解异步确认是怎么实现的。
//异步发布确认 public static void publishMessageAsync() throws Exception{ Channel channel = RabbitMqUtils.getChannel(); //队列的声明 String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName,true,false,false,null); //开启发布确认 channel.confirmSelect(); //开始时间 long begin = System.currentTimeMillis(); //消息确认成功，回调函数 ConfirmCallback ackCallback = (deliveryTag,multiple)->{ System.out.println("确认的消息:"+deliveryTag); }; //消息确认失败，回调函数 ConfirmCallback nackCallback = (deliveryTag,multiple)->{ System.out.println("未确认的消息:"+deliveryTag); }; //准备消息监听器 监听哪些消息成功了 哪些消息失败了 channel.addConfirmListener(ackCallback,nackCallback); //批量发送消息 for (int i = 0; i < MESSAGE_COUNT; i++) { String message = "消息"+i; channel.basicPublish("",queueName,null,message.getBytes()); } //结束时间 long end = System.currentTimeMillis(); System.out.println("发布"+MESSAGE_COUNT+"个确认消息，耗时"+(end-begin+"ms")); }
如何处理异步未确认消息
最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用ConcurrentLinkedQueue这个队列在confirm callbacks与发布线程之间进行消息的传递。
//异步发布确认 public static void publishMessageAsync() throws Exception{ Channel channel = RabbitMqUtils.getChannel(); //队列的声明 String queueName = UUID.randomUUID().toString(); channel.queueDeclare(queueName,true,false,false,null); //开启发布确认 channel.confirmSelect(); //线程安全有序的一个哈希表适用于高并发的情况下 /* * 1.轻松的将序号与消息进行关联 * 2.轻忪批量删除条目只要给到序号 3.支持高并发（多线程)*/ ConcurrentSkipListMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap<>(); //开始时间 long begin = System.currentTimeMillis(); //消息确认成功，回调函数 ConfirmCallback ackCallback = (deliveryTag,multiple)->{ if(multiple){ ConcurrentNavigableMap<Long,String> confirmed = outstandingConfirms.headMap(deliveryTag); confirmed.clear(); }else { outstandingConfirms.remove(deliveryTag); } //2.删除到已经确认的消息 剩下的就是未确认的消息 System.out.println("确认的消息:"+deliveryTag); }; //消息确认失败，回调函数 ConfirmCallback nackCallback = (deliveryTag,multiple)->{ //打印一下未确认的消息都有哪些 String message = outstandingConfirms.get(deliveryTag); System.out.println("未确认的消息:"+message+"::::"+deliveryTag); }; //准备消息监听器 监听哪些消息成功了 哪些消息失败了 channel.addConfirmListener(ackCallback,nackCallback); //批量发送消息 for (int i = 0; i < MESSAGE_COUNT; i++) { String message = "消息"+i; channel.basicPublish("",queueName,null,message.getBytes()); //1.此处记录下所有要发送的消息 消息的总和 outstandingConfirms.put(channel.getNextPublishSeqNo(),message); } //结束时间 long end = System.currentTimeMillis(); System.out.println("发布"+MESSAGE_COUNT+"个确认消息，耗时"+(end-begin+"ms")); }
交换机
在上一节中，我们创建了一个工作队列。我们假设的是工作队列背后，每个任务都恰好交付给一个消费者(工作进程)。在这一部分中，我们将做一些完全不同的事情-我们将消息传达给多个消费者。这种模式称为发布/订阅"。
概念
RabbitMQ消息传递模型的核心思想是:生产者生产的消息从不会直接发送到队列。实际上，通常生产者甚至都不知道这些消息传递传递到了哪些队列中。
相反，生产者只能将消息发送到交换机(exchange)，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列还是说把他们到许多队列中还是说应该丢弃它们。这就的由交换机的类型来决定。
交换机类型
直接（direct），主题（topic），标题（headers），扇出（fanout）
临时队列
每当我们连接到Rabbit时，我们都需要一个全新的空队列，为此我们可以创建一个具有随机名称的队列，或者能让服务器为我们选择一个随机队列名称那就更好了。其次一旦我们断开了消费者的连接，队列将被自动删除。
绑定
什么是 bingding呢，binding其实是exchange和queue之间的桥梁，它告诉我们exchange和那个队列进行了绑定关系。比如说下面这张图告诉我们的就是×与Q1和Q2进行了绑定
Fanout交换机
Fanout这种类型非常简单。正如从名称中猜到的那样，它是将接收到的所有消息广播到它知道的所有队列中。系统中默认有些exchange类型
Fannout交换机（消费者）
public class ReceiveLogs02 { //交换机的名称 public static final String EXCHANGE_NAME="logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //产生一个交换机 channel.exchangeDeclare(EXCHANGE_NAME,"fanout"); //声明一个队列 临时队列 //当消费者斯开与队列的连接的时候队列就自动鹏除 String queueName = channel.queueDeclare().getQueue(); //绑定交换机与 channel.queueBind(queueName,EXCHANGE_NAME,""); System.out.println("ReceiveLogs01等待接收消息，把接收到消息打印在屏幕上......"); //接收消息 DeliverCallback deliverCallback = (consumerTag,message)->{ System.out.println("控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8")); }; //消费者取消消息时回调接口 channel.basicConsume(queueName,true,deliverCallback,consumerTag->{}); } }
public class ReceiveLogs02 { //交换机的名称 public static final String EXCHANGE_NAME="logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //产生一个交换机 channel.exchangeDeclare(EXCHANGE_NAME,"fanout"); //声明一个队列 临时队列 //当消费者斯开与队列的连接的时候队列就自动鹏除 String queueName = channel.queueDeclare().getQueue(); //绑定交换机与 channel.queueBind(queueName,EXCHANGE_NAME,""); System.out.println("ReceiveLogs02等待接收消息，把接收到消息打印在屏幕上......"); //接收消息 DeliverCallback deliverCallback = (consumerTag,message)->{ System.out.println("控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8")); }; //消费者取消消息时回调接口 channel.basicConsume(queueName,true,deliverCallback,consumerTag->{}); } }
Fannout交换机（生产者）
//发消息 交换机 public class EmitLog { //交换机的名称 public static final String EXCHANGE_NAME = "logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //channel.exchangeDeclare(EXCHANGE_NAME,"fauout"); Scanner scanner = new Scanner(System.in); while (scanner.hasNext()){ String message = scanner.next(); channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("UTF-8")); System.out.println("生产者发出消息:"+message); } } }
运行结果：
Direct exchage(直接交换机)
上一节中的我们的日志系统将所有消息广播给所有消费者，对此我们想做一些改变，例如我们希望将日志消息写入磁盘的程序仅接收严重错误(erros)，而不存储哪些警告(warning)或信息(info)日志消息避免浪费磁盘空间。Fanout这种交换类型并不能给我们带来很大的灵活性-它只能进行无意识的广播，在这里我们将使用direct这种类型来进行替换，这种类型的工作方式是，消息只去到它绑定的routingKey队列中去。
生产者
public class DirectLogs { //交换机的名称 public static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //channel.exchangeDeclare(EXCHANGE_NAME,"fauout"); Scanner scanner = new Scanner(System.in); while (scanner.hasNext()){ String message = scanner.next(); channel.basicPublish(EXCHANGE_NAME,"info",null,message.getBytes("UTF-8")); System.out.println("生产者发出消息:"+message); } } }
消费者
public class ReceiveLogDirect01 { public static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); //声明一个队列 channel.queueDeclare("console", false,false,false,null); channel.queueBind("console",EXCHANGE_NAME,"info"); //接收消息 DeliverCallback deliverCallback = (consumerTag, message)->{ System.out.println("ReceiveLogDirect01控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8")); }; //消费者取消消息时回调接口 channel.basicConsume("console",true,deliverCallback,consumerTag->{}); } }
public class ReceiveLogDirect02 { public static final String EXCHANGE_NAME = "direct_logs"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明一个交换机 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); //声明一个队列 channel.queueDeclare("disk", false,false,false,null); channel.queueBind("disk",EXCHANGE_NAME,"error"); //接收消息 DeliverCallback deliverCallback = (consumerTag, message)->{ System.out.println("ReceiveLogDirect02控制台打印接收到的消息:"+new String(message.getBody(),"UTF-8")); }; //消费者取消消息时回调接口 channel.basicConsume("disk",true,deliverCallback,consumerTag->{}); } }
运行结果:
