Work Queues：与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。
应用场景：对于任务过重或任务较多情况，使用工作队列可以提高任务处理的速度。
ProducerWorkQueues
public class ProducerWorkQueues {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
// 创建队列Queue
channel.queueDeclare("work_queues", true, false, false, null);
Scanner scanner = new Scanner(System.in);
while(scanner.hasNext()){
String message = scanner.next();
// 发送消息
channel.basicPublish("", "work_queues", null, message.getBytes());
System.out.println("消息发送成功：" + message);
}
// 释放资源
channel.close();
connection.close();
}
}
ConsumerWorkQueues1
public class ConsumerWorkQueues1 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtil.getChannel();
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("消费者启动等待消费消息................");
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者接收到的消息:" + new String(body));
}
};
channel.basicConsume("work_queues", true, consumer);
}
}
ConsumerWorkQueues2
public class ConsumerWorkQueues1 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtil.getChannel();
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("消费者启动等待消费消息................");
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者接收到的消息:" + new String(body));
}
};
channel.basicConsume("work_queues", true, consumer);
}
}
先运行两个消费者，然后再运行生产者，运行结果如下：
1、在一个队列中，如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系。消费者之间按照有序的一个接收一次消息。
2、Work Queues：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如：短信服务部署多个，只需要有一个节点成功发送即可。
消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然挂掉了，会发生什么情况呢？RabbitMQ 一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息，以及后续发送给该消费者的消息。
为了保证消息在发送过程中不丢失，RabbitMQ 引入了消息应答机制，消息应答就是：消费者在接收到消息并且处理该消息之后，告诉 RabbitMQ 它已经处理了，RabbitMQ 可以把该消息删除了。
2.2 自动应答消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡，因为这种模式如果消息在接收到之前，消费者那边出现连接或者 channel 关闭，那么消息就丢失了，当然另一方面，这种模式消费者那边可以传递过载的消息，没有对传递的消息数据进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终是的内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。
消费者在订阅队列时，可以指定 autoAck 参数，当 autoAck 等于 false 时，RabbitMQ 会等待消费者显示地恢复确认信号后才从内存（或者磁盘）中移去消息（实质上是先打上删除标记，之后再删除）。当 autoAck 等于 true 时，RabbitMQ 会自动把发送出去的消息置为确认，然后从内存（或磁盘）中删除，而不管消费者是否真正地消费到了这些消息。
采用消息应答机制后，只要设置 autoAck 参数为 false，消费者就有足够的时间处理消息（任务），不用担心处理消息过程中消费者进程挂掉后消息丢失的问题，因为 RabbitMQ 会一直等待持有消息知道消费者显式调用 Basic.Ack 命令为止。
当 autoAck 参数置为 false，对于 RabbitMQ 服务端而言，队列中的消息分成了两个部分：一部分是等待投递给消费者的消息；一部分是已经投递给消费者，但是还没有收到消费者确认信号的消息。如果 RabbitMQ 一直没有收到消费者的确认信号，并且消费此消息的消费者已经断开连接，则 RabbitMQ 会安排该消息重新进入队列，等待投递给下一个消费者，当然也有可能还是原来的那个消费者。
RabbitMQ 不会为未确认的消息设置过期时间，它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息的消费者连接是否已经断开，这么设计的原因是 RabbitMQ 允许消费者消费一条消息的时间可以很久很久。
2.3 消息应答的方法- Channel.basicAck（用于肯定确认），RabbitMQ 已经知道该消息并且成功的处理消息，可以将其丢弃了
- Channel.basicNack（用于否定确认）
- Channel.basicReject（用于否定确认），与 Channel.basicNack 相比少一个参数。不处理该消息了直接拒绝，可以将其丢弃了。
手动应答的好处是可以批量应答并且减少网络拥堵
multiple 的 true 和 false 代表不同的意思：
- true 代表批量应答 channel 上未应答的消息，比如说 channel 上有传送 tag 的消息 5、6、7、8，当前 tag 是 8，那么此时 5~8 的这些还未应答的消息都会被确认收到消息应答。
- false 同上面相比，只会应答 tag = 8 的消息，5、6、7这三个消息依然不会被确认收到消息应答
如果消费者由于某种原因失去连接（其通道已关闭，连接已关闭或 TCP 连接丢失），导致消息未发送 ACK 确认，RabbitMQ 将连接到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。
默认消息用的是自动应答，所以我们要想实现消息消费过程中不丢失，需要把自动应答改为手动应答。
生产者：ProducerWorkQueues2
public class ProducerWorkQueues2 {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
// 创建队列Queue
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("等待消息输入：");
Scanner scanner = new Scanner(System.in);
while(scanner.hasNext()){
String message = scanner.next();
// 发送消息
channel.basicPublish("", "work_queues", null, message.getBytes());
System.out.println("消息发送成功：" + message);
}
// 释放资源
channel.close();
connection.close();
}
}
消费者：WorkQueues1
public class WorkQueues1 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtil.getChannel();
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("消费者启动等待消费消息................处理时间较短");
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
SleepUtils.sleep(1);
System.out.println("消费者接收到的消息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
boolean autoAck = false;
channel.basicConsume("work_queues", autoAck, consumer);
}
}
消费者：WorkQueues2
public class WorkQueues2 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtil.getChannel();
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("C2消费者启动等待消费消息................处理时间较长");
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
SleepUtils.sleep(3000);
System.out.println("消费者接收到的消息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
boolean autoAck = false;
channel.basicConsume("work_queues", autoAck, consumer);
}
}
SleepUtils
public class SleepUtils {
public static void sleep(int second){
try {
Thread.sleep(1000*second);
} catch (InterruptedException _ignored) {
Thread.currentThread().interrupt();
}
}
}
启动生产者和两个消费者，通过生产者发送两条消息
消息发送完毕后，将消费者 WorkQueue2 停止运行。
在生产者发送消息 bb，发出消息之后的把 C2 消费者停掉，按理说该 C2 来处理该消息，但是由于它处理时间较长，在还未处理完，也就是说 C2 还没有执行 ack 代码的时候，C2 被停掉了，此时会看到消息被 C1 接收到了，说明消息 bb 被重新入队，然后分配给能处理消息的 C1 处理了。
public class WorkQueues2 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtil.getChannel();
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("消费者启动等待消费消息................处理时间较长");
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者接收到的消息:" + new String(body));
channel.basicNack(envelope.getDeliveryTag(), false, false);
}
};
boolean autoAck = false;
channel.basicConsume("work_queues", autoAck, consumer);
}
}
2.6.2.2 basicReject 方法
public class WorkQueues1 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtil.getChannel();
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("消费者启动等待消费消息................处理时间较短");
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
SleepUtils.sleep(1);
System.out.println("消费者接收到的消息:" + new String(body));
channel.basicReject(envelope.getDeliveryTag(), false);
}
};
boolean autoAck = false;
channel.basicConsume("work_queues", autoAck, consumer);
}
}
Basic.Reject 命令一次只能拒绝一条消息，Basic.Nack 可以批量拒绝消息。
如果要求处理消息速度快，高吞吐量，使用自动应答。
如果对于数据安全性有要求，使用手动应答，可以保证消息不会丢失。
默认情况下，RabbitMQ 退出或者由于某种原因崩溃时，它忽视队列和消息，除非告知它不要这样做。确保消息不会丢失需要做两件事：需要将队列话消息都标记为持久化。
3.2 队列实现持久化之前我们创建的队列都是非持久化的，RabbitMQ 如果重启的话，该队列就会被删除掉，如果要队列实现持久化，需要在声明队列的时候把 durable 参数设置为持久化。
但是需要注意的就是，如果之前声明的队列不是持久化的，需要把原先队列先删除，或者重新创建一个持久化的队列，不然就会出现错误。
以下为控制台中持久化与非持久化的 UI 显示区：
非持久化：
持久化：
这个时候即使重启 RabbitMQ，队列也依然存在。
3.3 消息实现持久化要想让消息实现持久化，需要在消息生产者修改代码，如下所示，添加属性 MessageProperties.PERSISTENT_TEXT_PLAIN，要想该属性起作用，需要先保证队列是持久化的。
将消息标记为持久化并不能完全保证不会丢失消息。当消息刚准备存储在磁盘的时候，此时 RabbitMQ 发生宕机，但是消息并未存储完，这就有可能造成消息丢失。因而，持久性保证并不强，但是对于简单任务队列而言，这已经绰绰有余了。如果需要更强有力的持久化策略，可以参考“发布确认”。
前面讲到，RabbitMQ 在分发消息时采用的是轮询分发机制，但是在某种场景下这种策略并不是很好，比方说有两个消费者在处理任务，其中消费者 1 处理任务的速度非常快，而消费者 2 处理速度却很慢，这个时候如果我们还是采用轮询分发的话，那么消费者 1 就会有很大一部分时间处于空闲状态，而消费者 2 则一直在干活。
为了避免这种情况，我们可以设置参数 channel.basicQos(1);
由于消息的发送是异步发送的，所以在任何时候，channel 上肯定不止只有一条消息，另外，来自消费者的手动确认本质上也是异步的。因此，这里就存在一个未确认的消息缓冲区，那么可以通过限制此缓冲区的大小，来避免缓冲区里面无限制的未确认消息的问题。这个时候就可以通过使用 basicQos 方法设置“预取计数：值来完成。该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量，RabbitMQ 将停止在通道上传递更多消息，除非至少有一个未处理的消息被确认。例如，假设在通道上有未确认的消息 5、6、7、8，并且通道的预取计数设置为 4，此时 RabbitMQ 将不会在该通道上再传递任何消息，除非至少有一个未应答的消息被 ack。比方说 tag = 6 的这条消息刚刚被确认 ack，RabbitMQ 将会感知到这个情况并在发送一条消息。消息应答和 QoS 预取值对用户吞吐量有重大影响。通常，增加预取将提高向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息的数量也会增加，从而增加了消费者的 RAM 消耗（随机存取存储器），应该小心使用具有无限预处理的自动确认模式或手动确认模式，消费者消费了大量的消息如果没有确认的话，会导致消费者连接节点的内存消耗变大，所以找到合适的预取值是一个反复试验的过程，不同的负载改制取值也不同，100到300范围内的值通常可提供最佳的吞吐量，并且不会给消费者带来太大的风险。预取值为 1 是最保守的。当然这将使吞吐量变得很低，特别是消费者连接延迟很严重的情况下，特别是在消费者连接等待时间较长的环境中。对于大多数应用来说，稍微高一点的值将是最佳的。
生产者代码
public class ProducerWorkQueues2 {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
// 创建队列Queue
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("等待消息输入：");
Scanner scanner = new Scanner(System.in);
while(scanner.hasNext()){
String message = scanner.next();
// 发送消息
channel.basicPublish("", "work_queues", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
System.out.println("消息发送成功：" + message);
}
// 释放资源
channel.close();
connection.close();
}
}
消费者1 代码：
public class WorkQueues1 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtil.getChannel();
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("消费者启动等待消费消息................处理时间较短");
// 设置预取值 该值定义的是通道上未确认消息的最大个数
int prefetchCount = 3;
channel.basicQos(prefetchCount);
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
SleepUtils.sleep(1);
System.out.println("消费者接收到的消息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
boolean autoAck = false;
channel.basicConsume("work_queues", autoAck, consumer);
}
}
消费者 2 代码
public class WorkQueues2 {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtil.getChannel();
channel.queueDeclare("work_queues", true, false, false, null);
System.out.println("消费者启动等待消费消息................处理时间较长");
// 设置预取值
int prefetchCount = 2;
channel.basicQos(prefetchCount);
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
SleepUtils.sleep(30);
System.out.println("消费者接收到的消息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
boolean autoAck = false;
channel.basicConsume("work_queues", autoAck, consumer);
}
}
通过设置 Thread.sleep 的时间长短来模拟不同消费者的处理消息的速度。分别运行生产者以及两个消费者，通过生产者生产 7 条消息，来观察两个消费者消费消息的情况：
生产者：
消费者 1 由于处理速度较快，且 basicQoS 设置的值为 3，消费了 5 条消息
消费者 2 由于处理速度较慢，且 basicQoS 设置的值为 2 ，小于消费者 1 的预取值，只消费了 2 条消息
