工作队列
逻辑图
<!-- SpringBoot 消息队列的起步依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
轮询分发 Round-robin
生产者
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeoutException;
public class WorkQueueProducer {
/**
* 生产者 → 消息队列
* 创建连接工厂,并设置参数
* 创建连接 Connection
* 创建通道 Channel
* 创建队列
* 发送消息
**/
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
if(true){
factory.setHost("127.0.0.1");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
}
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//创建队列
/**
* String queue 队列名称
* boolean durable 是否持久化，
* boolean exclusive 含义一:是否独占,是否只能有一个消费者监听
* 含义二:connection 关闭是否删除队列
* boolean autoDelete 是否自动删除,当没有消费者的时候是否自动删除
* Map<String, Object> arguments 参数
*/
channel.queueDeclare("WorkQueues",true,false,false,null);
//发送消息
/**
* String exchange : 交换机名称,简单模式不使用交换机
* String routingKey : 路由规则,当不使用交换机时,路由键需要和队列名称相同
* BasicProperties props : 配置参数
* byte[] body : 消息体,真实的数据
*/
for (int i = 0; i < 20; i++) {
String str = "WorkQueues is so easy!\t" + i + "\t" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());;
System.out.println("发送消息:\t" + str);
channel.basicPublish("","WorkQueues",null,str.getBytes());
}
//释放资源
channel.close();
connection.close();
System.out.println("消息发送成功");
}
}
- 与简单队列几乎没有什么不同
消费者
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class WorkQueueConsumerA {
/**
* 消息队列 ← 消费者
* 创建连接工厂,并设置参数
* 创建连接 Connection
* 创建通道 Channel
* 订阅队列
* 接收消息
*/
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接工厂,并设置参数
ConnectionFactory factory = new ConnectionFactory();
if (true) {
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
}
//创建连接 Connection
Connection connection = factory.newConnection();
//创建通道 Channel
Channel channel = connection.createChannel();
/**
* consumerTag 消费信息标签
* delivery 回执
*/
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
byte[] body = delivery.getBody();
System.out.println("消费消息:\t" + new String(body));
};
/**
* basicConsume(String queue, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback)
* String queue : 队列名称
* boolean autoAck : 是否自动确认,如果true,消费者接收到消息会自动发送一个回执给消息队列
* DeliverCallback deliverCallback : 回调函数
* CancelCallback cancelCallback : 消费者取消订阅时的回调函数
*/
channel.basicConsume("WorkQueues", true, deliverCallback, consumerTag -> {
});
}
}
- 再创建一个类 WorkQueueConsumerB，代码与 WorkQueueConsumerA 一样，只是类型不同
测试
-
先启动生产者，查看 RabbitMQ 网页控制台
-
先将2个消费者启动
- 第一个消费者启动的时候，会将所有的都消费掉
- 将两个都启动之后，再启动生产者
-
再启动生产者
公平分发 Fair
如果机器 A 性能很好，一下子就处理完了，其他时间一直空闲，而机器 B 性能很差，很久都不能处理完一条，但是队列还是一人一条的轮询分发，这就造成 A 性能浪费，B 处理慢
我们采用公平分发
采用 basicQos(prefetchCount=1) ，来限制 MQ 只发不超过1条的消息给同一个消费者，当消费者处理完消息，给 MQ 反馈了，MQ 才会进行第二次发送
生产者
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeoutException;
public class WorkQueueProducer {
/**
* 生产者 → 消息队列
* 创建连接工厂,并设置参数
* 创建连接 Connection
* 创建通道 Channel
* 创建队列
* 发送消息
**/
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
if(true){
factory.setHost("127.0.0.1");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
}
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//创建队列
/**
* String queue 队列名称
* boolean durable 是否持久化，
* boolean exclusive 含义一:是否独占,是否只能有一个消费者监听
* 含义二:connection 关闭是否删除队列
* boolean autoDelete 是否自动删除,当没有消费者的时候是否自动删除
* Map<String, Object> arguments 参数
*/
channel.queueDeclare("WorkQueues",true,false,false,null);
//发送消息
/**
* String exchange : 交换机名称,简单模式不使用交换机
* String routingKey : 路由规则,当不使用交换机时,路由键需要和队列名称相同
* BasicProperties props : 配置参数
* byte[] body : 消息体,真实的数据
*/
for (int i = 0; i < 20; i++) {
String str = "WorkQueues is so easy!\t" + i + "\t" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());;
System.out.println("发送消息:\t" + str);
channel.basicPublish("","WorkQueues",null,str.getBytes());
}
//释放资源
channel.close();
connection.close();
System.out.println("消息发送成功");
}
}
消费者
消费者A
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class WorkQueueConsumerA {
/**
* 消息队列 ← 消费者
* 创建连接工厂,并设置参数
* 创建连接 Connection
* 创建通道 Channel
* 订阅队列
* 接收消息
*/
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接工厂,并设置参数
ConnectionFactory factory = new ConnectionFactory();
if (true) {
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
}
//创建连接 Connection
Connection connection = factory.newConnection();
//创建通道 Channel
Channel channel = connection.createChannel();
/**
* prefetchCount 设为 1
* MQ 发送小于等于 1 的数据给消费者
* 当消费者消费完这几条数据,就会给 MQ 一个反馈,MQ 再次发送
*/
channel.basicQos(1);
/**
* consumerTag 消费信息标签
* delivery 回执
*/
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
byte[] body = delivery.getBody();
System.out.println("消费消息:\t" + new String(body));
try {
/**
* 睡眠 1 秒,模拟等待
*/
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
/**
* 手动回执
* long deliveryTag
* boolean multiple
*/
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
};
/**
* basicConsume(String queue, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback)
* String queue : 队列名称
* boolean autoAck : 是否自动确认,如果true,消费者接收到消息会自动发送一个回执给消息队列
* DeliverCallback deliverCallback : 回调函数
* CancelCallback cancelCallback : 消费者取消订阅时的回调函数
*/
channel.basicConsume("WorkQueues", false, deliverCallback, consumerTag -> {
});
}
}
消费者B
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class WorkQueueConsumerB {
/**
* 消息队列 ← 消费者
* 创建连接工厂,并设置参数
* 创建连接 Connection
* 创建通道 Channel
* 订阅队列
* 接收消息
*/
public static void main(String[] args) throws IOException, TimeoutException {
//创建连接工厂,并设置参数
ConnectionFactory factory = new ConnectionFactory();
if (true) {
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
}
//创建连接 Connection
Connection connection = factory.newConnection();
//创建通道 Channel
Channel channel = connection.createChannel();
/**
* prefetchCount 设为 1
* MQ 发送小于等于 1 的数据给消费者
* 当消费者消费完这几条数据,就会给 MQ 一个反馈,MQ 再次发送
*/
channel.basicQos(1);
/**
* consumerTag 消费信息标签
* delivery 回执
*/
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
byte[] body = delivery.getBody();
System.out.println("消费消息:\t" + new String(body));
try {
/**
* 睡眠 1 秒,模拟等待
*/
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
/**
* 手动回执
* long deliveryTag
* boolean multiple
*/
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
};
/**
* basicConsume(String queue, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback)
* String queue : 队列名称
* boolean autoAck : 是否自动确认,如果true,消费者接收到消息会自动发送一个回执给消息队列
* DeliverCallback deliverCallback : 回调函数
* CancelCallback cancelCallback : 消费者取消订阅时的回调函数
*/
channel.basicConsume("WorkQueues", false, deliverCallback, consumerTag -> {
});
}
}
-
与轮询的区别
-
channel.basicQos(1); -
try { /** * 睡眠 1 秒,模拟等待 */ TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } -
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); -
自动确认的 autoAck 改为 false channel.basicConsume("WorkQueues", false, deliverCallback, consumerTag -> {});
-
测试
SpringBoot整合
小结
2 个消费者监听同一个队列，消息被平均分配到 2 个消费者，提高了处理效率，3个4个消费者效率更高
轮询分发：假设有100条消息，A消费者消费50条，B消费者消费50条，但是 A 机器是8核32G的，B机器是1核1G的，显然 B机器消费慢，A机器一直空闲
公平分发：性能好的机器多消费一点，性能差的少消费一点，负载均衡
![[数据集][目标检测]人脸口罩佩戴目标检测数据集VOC+YOLO格式8068张3类别](https://i-blog.csdnimg.cn/direct/088e80f82eb14e728a652ec9ba9fc6d4.png)
