Java RabbitMQ的工作队列与消息应答详解
作者：江海i
Work Queues
工作队列（任务队列）主要思想是避免立即执行资源密集型任务，而不得不等待它完成，相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。
其实就是生产者发送大量的消息，发送到队列之后，由多个消费者(工作线程)来处理消息，并且每个消息只能被处理一次。
1. 轮询分发消息
多个工作线程按照次序每来一个消息执行一次。
1.1 抽取工具类
直接通过信息获取信道
/**
* @Description RabbitMQ工具类
* @date 2022/3/5 10:02
*/
public class RabbitMQUtils {
public static Channel getChannel() throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("1");
factory.setUsername("guest");
factory.setPassword("guest");
Connection connection = factory.newConnection();
return connection.createChannel();
}
}
1.2 编写两个工作线程
Work2和Work1代码没有区别，只需要对它做出区分即可。
public class Worker1 {
// 指定队列名称
private static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
// 获取信道
Channel channel = RabbitMQUtils.getChannel();
// 声明：接收消息回调
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("工作线程01："+ new String(message.getBody()));
};
// 声明：取消消费回调
CancelCallback cancelCallback = consumerTag -> {
System.out.println("工作线程01取消接收："+consumerTag);
};
System.out.println("工作线程01启动完成......");
channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);
}
}
1.3 编写生产者
public class Producer {
private static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQUtils.getChannel();
// 产生队列
channel.queueDeclare(QUEUE_NAME,false,false,true,null);
// 消息体
Scanner scanner = new Scanner(System.in);
int i = 1;
while (scanner.hasNext()){
String msg = scanner.next();
msg = msg + i;
channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
System.out.println("发送成功：" + msg);
}
System.out.println("----------==========发送完毕==========----------");
}
}
1.4 运行测试
先启动两个工作线程，再启动生产者。
出现404异常请参考下方1.6
生产者发送情况：
轮询状态下两个工作队列接收状态：
1.5 异常情况
在先启动两个消费者线程时，会提示404找不到队列。
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no queue 'work_queue' in vhost '/', class-id=60, method-id=20)
发生这个情况的原因很显然是因为先启动了消费者，但是在RabbitMQ中没有创建相对应的队列名称，解决方法可以：
1.先启动生产者创建队列（也可以在RabbitMQ中创建队列）；
2.再启动消费者就不会产生这个错误；
3.再在生产者中使用Scanner类去发送消息测试。
2. 消息应答
消费者在接收到消息并且处理该消息之后，告诉RabbitMQ它已经处理了，RabbitMQ可以删除消息。其目的就是为了保护消息在被处理之前不会消失。
2.1 自动应答
这种方式发送后就被认定为已经传送成功，所以在消息接收到之前消费者的连接或者channel关闭，那么这个消息就会丢失。其特点是消费者可以传递过载的消息，对传递的消息没有限制，但如果因内存耗尽消费者线程被系统杀死，就会使得多条消息丢失。所以这个模式需要在数据安全性和吞吐量之间选择，适合使用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。
所以自动应答的方式局限性很高。
2.2 手动应答
优点：可以批量应答和减少网络拥挤。
1.channel.basicAck(long deliveryTag, boolean multiple);：肯应应答，处理完消息之后提醒RabbitMQ可以删除当前队列，deliveryTag：当前队列中选中的消息；multiple：是否批量应答。
2.channel.basicNack(long deliveryTag, boolean multiple, boolean requeue)：否定应答，
3.channel.basicReject(long deliveryTag, boolean requeue)：否定并且拒绝应答。
2.3 消息自动重新入队
如果消费者因为一些原因失去了对RabbitMQ的连接，导致没有发送ACK确认，RabbitMQ就会对该消息进行重新排队，并且分发给可以处理该消息的消费者，所以即使某个消费者死亡，也可以保证消息不会丢失。
2.4 手动应答测试
测试目的：在手动应答状态下不会发生消息丢失的情况。
测试方法：
1.创建两个消费者；
2.使用工具类使线程睡眠一定时间；
3.在睡眠时关闭线程，看能否自动重新入队。
2.4.1 生产者代码
/**
* @Description 手动应答生产者
* @date 2022/3/5 19:03
*/
public class Producer1 {
// 指定队列名
private static final String TASK_QUEUE_RES = "queue_res";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQUtils.getChannel();
channel.queueDeclare(TASK_QUEUE_RES,false,false,false,null);
Scanner scanner = new Scanner(System.in);
int i = 0;
while (scanner.hasNext()){
i++;
String msg = scanner.next() + i;
channel.basicPublish("",TASK_QUEUE_RES,null,msg.getBytes(StandardCharsets.UTF_8));
System.out.println("发送消息：'" + msg + "'成功");
}
}
}
2.4.2 消费者代码
/**
* @Description 手动应答消费者1
* @date 2022/3/5 19:17
*/
public class Worker1 {
private static final String TASK_QUEUE_RES = "queue_res";
public static void main(String[] args) throws Exception{
Channel channel = RabbitMQUtils.getChannel();
System.out.println("线程A等待接收......");
DeliverCallback deliverCallback = (consumerTag, message) -> {
// 模拟并发沉睡一秒
try {
Thread.sleep(1000);
System.out.println("线程A接收消息："+ new String(message.getBody(), StandardCharsets.UTF_8));
/**
* basicAck:
* 1. 消息标记
* 2. 是否批量
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
};
channel.basicConsume(TASK_QUEUE_RES,false,deliverCallback,
consumerTag -> {
System.out.println(consumerTag + "消费者取消消费");
});
}
}
Worker2类和1区别不大，将名称改成B再将睡眠事件改成30即可。
2.4.3 测试
测试方法：
1.先启动生产者创建队列；
2.启动两个消费者接收消息；
3.因为是轮询方式，所以A线程接收之后肯定是B线程接收，在睡眠时关闭B线程，如果A线程接收到说明测试成功。
发送消息：
线程A接收：
再发送消息：
关闭线程B线程A接收到消息：
测试成功！
总结
本篇文章就到这里了，希望能够给你带来帮助，也希望您能够多多关注脚本之家的更多内容!