2工作队列

工作队列

逻辑图

<!-- 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("/");}//创建连接 ConnectionConnection connection = factory.newConnection();//创建通道 ChannelChannel 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("/");}//创建连接 ConnectionConnection connection = factory.newConnection();//创建通道 ChannelChannel 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("/");}//创建连接 ConnectionConnection connection = factory.newConnection();//创建通道 ChannelChannel 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机器一直空闲

公平分发：性能好的机器多消费一点，性能差的少消费一点，负载均衡