工作模型（Work Queues）
- 一、轮询发送消息
- 二、消息应答
- 2.1、概念
- 2.2、自动应答
- 2.3、手动应答
- 2.4、消息的重新入队
- 2.5、手动应答代码
概念：工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。
相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进
程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。
通俗的理解：当大量的消息发送到MQ的时候，这个时候为了快速的消费这些消息，就会开启多个工作线程进行消费，为了消息不会重复的消费，所以采用轮询的方式。
示例图
一、轮询发送消息
创建工作线程
package com.feng.workQueue;
import com.feng.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Delivery;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Author Feng
* @Date 2022/11/16 16:09
* @Version 1.0
* @Description 工作线程 1号（也就是一号消费者）
*/
public class Worker01 {
//队列名
public static final String QUEUE_NAME = "hello";
/**
* 消费消息
*
* @param args
*/
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMQUtil.getChannel();
System.out.println("工作线程1等待消费。。。。。。。");
//consumerTag: 消费者的标签
channel.basicConsume(QUEUE_NAME,true,(consumerTag, message)->{
System.out.println("接收的消息是:"+new String(message.getBody()));
},consumerTag->{
System.out.println(consumerTag+"消息接收失败");
});
}
}
这里工作线程都是一样的，所以你可以自己开多线程或者直接使用IDEA自带的开启类的多个实例，如下
效果如下
生产者
package com.feng.workQueue;
import com.feng.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
/**
* @Author Feng
* @Date 2022/11/16 16:37
* @Version 1.0
* @Description 生产者代码
*/
public class Task01 {
//队列名
public static final String QUEUE_NAME = "hello";
//发送大量消息
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMQUtil.getChannel();
//队列的声明
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//从控制台进行输入消息进行发送
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String msg = scanner.next();
channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
System.out.println("消息发送成功"+msg);
}
}
}
测试
发送消息
工作线程1
工作线程2
由上可知工作线程的轮询
二、消息应答
2.1、概念
为了保证消息在发送过程中不丢失，rabbitmq 引入消息应答机制，消息应答就是:消费者在接收到消息并且处理该消息之后，告诉 rabbitmq 它已经处理了，rabbitmq 可以把该消息删除了
也就是在消费者进行消费处理完后会进行一个应答信号给MQ，然后MQ才可以将消息进行删除，防止消费者突然宕机所引发的消息丢失
2.2、自动应答
队列中的消息发送给消费者后立即被认为已经传送成功，然后删除消息
这种模式就是可以在消息发送就认为成功，对于提升处理大量消息的效率是很不错的有很大的吞吐量，但是缺点就是一旦消费者挂了，就会造成消息的丢失，或者是没有限制传递消息的数量，引起消息的堆积导致内存炸了，会将工作线程杀死
适用情况：在消费者可以高效并以某种速率能够处理这些消息的情况下使用。、
2.3、手动应答
// RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了
Channel.basicAck(用于肯定确认)
Channel.basicNack(用于否定确认)
// 与 Channel.basicNack 相比少一个参数 multiple(批量应答)
// 不处理该消息了直接拒绝，可以将其丢弃了
Channel.basicReject(用于否定确认)
Multiple 的解释
手动应答的好处是可以批量应答并且减少网络拥堵
-
true 代表批量应答 channel 上未应答的消息
比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是8 那么此时 5-8 的这些还未应答的消息都会被确认收到消息应答 -
false 同上面相比
只会应答 tag=8 的消息 5,6,7 这三个消息依然不会
也就是一批消息都处理完了只需要应答一个就行了，不需要每一个消息都进行应答确认，提高效率
2.4、消息的重新入队
如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或 TCP 连接丢失)，导致消息未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。
这里有个问题。就是消费者消费消息后，发生异常，ACK失败的话，消息会重新入队，这个消息会不会重新消费，为了防止这个问题，我们可以在消费者的程序里捕获异常，防止ACK失败后对消息进行消费，并且记录下这个消息以及错误，还有一种情况就是消费端都挂了，他就会不停的重新入队，会造成mq的不断执行同一个操作，我们可以设置重新入队的次数限制
2.5、手动应答代码
生产者
package com.feng.shouDongYingDa;
import com.feng.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;
/**
* @Author Feng
* @Date 2022/11/24 13:38
* @Version 1.0
* @Description 手动应答测试生产者
*/
public class Task2 {
public static final String TASK_QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMQUtil.getChannel();
channel.queueDeclare(TASK_QUEUE_NAME,false,false,false,null);
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String msg = scanner.next();
channel.basicPublish("",TASK_QUEUE_NAME,null,msg.getBytes("UTF-8"));
System.out.println("生产者发出消息:"+msg);
}
}
}
消费者1（处理消息时间较短）
package com.feng.shouDongYingDa;
import com.feng.utils.RabbitMQUtil;
import com.feng.utils.SleepUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Delivery;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Author Feng
* @Date 2022/11/24 13:47
* @Version 1.0
* @Description 手动应答消费者1(处理消息时间较短的)
*/
public class Work03 {
public static final String TASK_QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMQUtil.getChannel();
System.out.println("C1消费消息的时间较短");
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME,autoAck,(String consumerTag, Delivery message)->{
//睡一秒，这里用了一个自己封装的工具类，就是毫秒转秒
SleepUtils.sleep(1);
System.out.println("接收到的消息是:"+new String(message.getBody(),"UTF-8"));
//手动应答
/**
* 1.消息的标记 (Envelopes就是一些消息属性的封装类)
* 2.是否批量应答
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
},consumerTag->{
});
}
}
消费者2（处理消息时间较长）
package com.feng.shouDongYingDa;
import com.feng.utils.RabbitMQUtil;
import com.feng.utils.SleepUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Delivery;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Author Feng
* @Date 2022/11/24 13:47
* @Version 1.0
* @Description 手动应答消费者2(处理消息时间较长的)
*/
public class Work04 {
public static final String TASK_QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMQUtil.getChannel();
System.out.println("C2消费消息的时间较长");
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME,autoAck,(String consumerTag, Delivery message)->{
//睡一秒
SleepUtils.sleep(30);
System.out.println("接收到的消息是:"+new String(message.getBody(),"UTF-8"));
//手动应答
/**
* 1.消息的标记 (Envelopes就是一些消息属性的封装类)
* 2.是否批量应答
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
},consumerTag->{
});
}
}
测试
因为是轮询消费消息的，所以发两个消息是一个消费者一个，但是因为消费者2睡了30s，所以比较慢，如果在其睡眠时间中断程序模拟宕机，消息并不会丢失，而是会重新入队，由其他消费者消费，如下所示
