Work模式与前面的Simple模式一致,也是消息经由生产者发到Exchange再到queue再被消费者消费。不同点在于SIMPL模式是一个队列对应的一个消费者,此模式会由一个队列对应两个消费者或大于两个消费者。
此模式下,我们的消费者实现了对消息的平均消费,但如果消费者1消费能力若于消费者2,那就会造成消费者1 的消息积压,这时候我们就会想到使用公平模式,能者多劳,消费快的多消费,消费少的少消费。合理消息。后面代码会涉及到。
我们还是举小明洗澡的例子,小明洗澡比较孤独,这时候小明的女朋友小丽也来一块洗澡了,但两个人不想公用一个喷头,便又接入了一个喷头,一人一个开始洗,但是小明洗澡比较快,虽然洗完了,但是喷头关不上,只好把水浪费掉,小丽洗澡比较慢,但感觉水又不太够用。一共就20升水,一人10升。
2.1.1 生产者
-
- /**
- * @author rabbit
- * @version 1.0.0
- * @Description 一个生产者,一个默认的交换机,一个队列,两个消费者
- * @createTime 2022/07/27 19:34:00
- */
- public class WaterProducer {
-
- public static final String QUEUE_NAME = "SolarWaterHeater";
-
- //生产者
- public static void main(String[] args) throws Exception {
- //1、获取connection
- Connection connection = RabbitCommonConfig.getConnection();
- //2、创建channel
- Channel channel = connection.createChannel();
-
- for (int i = 1; i <= 20; i++) {
- sendMsg(channel, i);
- Thread.sleep(200);
- }
- //4、关闭管道和连接
- channel.close();
- connection.close();
- }
-
- private static void sendMsg(Channel channel, int k) throws IOException {
- //3、发送消息到exchange
- String msg = k + "升";
- /**
- * 参数1:指定exchange,使用“”。默认的exchange
- * 参数2:指定路由的规则,使用具体的队列名称。exchange为""时,消息直接发送到队列中
- * 参数3:指定传递的消息携带的properties
- * 参数4:指定传递的消息,byte[]类型
- */
- channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
- System.out.println("水龙头放水成功!" + k + "升");
- }
-
- }
2.1.2 消费者
小明洗澡:
-
- /**
- * @author rabbit
- * @version 1.0.0
- * @Description 工作模式 一个生产者,一个默认的交换机,一个队列,两个消费者
- * 需要在consumer消费者端,平均分配
- * @createTime 2022/07/27 19:36:00
- */
- public class XMShowerConsumer {
-
- public static final String QUEUE_NAME = "SolarWaterHeater";
-
- //消费者
- public static void main(String[] args) throws Exception {
- //1、获取连对象、
- Connection connection = RabbitCommonConfig.getConnection();
- //2、创建channel
- Channel channel = connection.createChannel();
- //3、创建队列-helloworld
- /**
- * 参数1:queue 指定队列名称
- * 参数2:durable 是否开启持久化(true)
- * 参数3:exclusive 是否排外(当前队列只能被一个消费者消费)
- * 参数4:autoDelete 如果这个队列没有其他消费者在消费,队列自动删除
- * 参数5:arguments 指定队列携带的信息
- */
- channel.queueDeclare(QUEUE_NAME, true, false, false, null);
-
- //4.开启监听Queue
- DefaultConsumer consumer = new DefaultConsumer(channel) {
- int i = 1;
-
- @Override
- public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
- try {
- Thread.sleep(500);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- if (i > 5) {
- System.out.println("小明洗澡洗完,开始浪费-第:" + new String(body, "UTF-8"));
- } else {
- System.out.println("小明洗澡已用水-第: " + new String(body, "UTF-8"));
- }
- i++;
- }
- };
- /**
- * 参数1:queue 指定消费哪个队列
- * 参数1:deliverCallback 指定是否ACK(true:收到消息会立即告诉RabbiMQ,false:手动告诉)
- * 参数1:cancelCallback 指定消费回调
- */
- channel.basicConsume(QUEUE_NAME, true, consumer);
- System.out.println("小明开始洗澡-快速洗......");
-
- //5、键盘录入,让程序不结束!
- System.in.read();
-
- //6、释放资源
- channel.close();
- connection.close();
- }
-
- }
小丽洗澡:
-
- /**
- * @author rabbit
- * @version 1.0.0
- * @Description 工作模式 一个生产者,一个默认的交换机,一个队列,两个消费者
- * 需要在consumer消费者端,平均分配
- * @createTime 2022/07/27 19:36:00
- */
- public class XLShowerConsumer {
-
- public static final String QUEUE_NAME = "SolarWaterHeater";
-
- //消费者
- public static void main(String[] args) throws Exception {
- //1、获取连对象、
- Connection connection = RabbitCommonConfig.getConnection();
- //2、创建channel
- Channel channel = connection.createChannel();
- //3、创建队列-helloworld
- /**
- * 参数1:queue 指定队列名称
- * 参数2:durable 是否开启持久化(true)
- * 参数3:exclusive 是否排外(当前队列只能被一个消费者消费)
- * 参数4:autoDelete 如果这个队列没有其他消费者在消费,队列自动删除
- * 参数5:arguments 指定队列携带的信息
- */
- channel.queueDeclare(QUEUE_NAME, true, false, false, null);
-
- //4.开启监听Queue
- DefaultConsumer consumer = new DefaultConsumer(channel) {
- @Override
- public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
- try {
- Thread.sleep(2000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- System.out.println("小丽洗澡已用水-第: " + new String(body, "UTF-8"));
- }
- };
- /**
- * 参数1:queue 指定消费哪个队列
- * 参数1:deliverCallback 指定是否ACK(true:收到消息会立即告诉RabbiMQ,false:手动告诉)
- * 参数1:cancelCallback 指定消费回调
- */
- channel.basicConsume(QUEUE_NAME, true, consumer);
- System.out.println("小丽开始洗澡-慢慢洗......");
-
- //5、键盘录入,让程序不结束!
- System.in.read();
-
- //6、释放资源
- channel.close();
- connection.close();
- }
-
- }
2.1.3 结果验证
生产者放水消息:
- 水龙头放水成功!1升
- 水龙头放水成功!2升
- 水龙头放水成功!3升
- 水龙头放水成功!4升
- 水龙头放水成功!5升
- 水龙头放水成功!6升
- 水龙头放水成功!7升
- 水龙头放水成功!8升
- 水龙头放水成功!9升
- 水龙头放水成功!10升
- 水龙头放水成功!11升
- 水龙头放水成功!12升
- 水龙头放水成功!13升
- 水龙头放水成功!14升
- 水龙头放水成功!15升
- 水龙头放水成功!16升
- 水龙头放水成功!17升
- 水龙头放水成功!18升
- 水龙头放水成功!19升
- 水龙头放水成功!20升
小明洗澡
- 小明开始洗澡-快速洗......
- 小明洗澡已用水-第: 2升
- 小明洗澡已用水-第: 4升
- 小明洗澡已用水-第: 6升
- 小明洗澡已用水-第: 8升
- 小明洗澡已用水-第: 10升
- 小明洗澡洗完,开始浪费-第:12升
- 小明洗澡洗完,开始浪费-第:14升
- 小明洗澡洗完,开始浪费-第:16升
- 小明洗澡洗完,开始浪费-第:18升
- 小明洗澡洗完,开始浪费-第:20升
小丽洗澡
- 小丽洗澡已用水-第: 1升
- 小丽洗澡已用水-第: 3升
- 小丽洗澡已用水-第: 5升
- 小丽洗澡已用水-第: 7升
- 小丽洗澡已用水-第: 9升
- 小丽洗澡已用水-第: 11升
- 小丽洗澡已用水-第: 13升
- 小丽洗澡已用水-第: 15升
- 小丽洗澡已用水-第: 17升
- 小丽洗澡已用水-第: 19升
从结果来看,两个人都是平均用水,虽然有快有慢,慢的就会存在堆积情况。
小丽就有意见了,自己不够洗,小明还白白浪费那么多水。小明给出了建议,让小丽把碰头开关开大点,自己开小点,这样就让小丽多用点水,自己少用点也就够了。
2.2.1 生产者
同上面生产者一致
2.2.2 消费者
小明洗澡:
-
- /**
- * @author rabbit
- * @version 1.0.0
- * @Description 工作模式 一个生产者,一个默认的交换机,一个队列,两个消费者
- * 需要在consumer消费者端,添加Qos能力以及更改为ACK手动即可让消费者根据自己的能力消费,不是RabbitMQ默认的平均分配了
- * @createTime 2022/07/27 19:36:00
- */
- public class XMShowerConsumer {
-
- public static final String QUEUE_NAME = "SolarWaterHeater";
-
- //消费者
- public static void main(String[] args) throws Exception {
- //1、获取连对象、
- Connection connection = RabbitCommonConfig.getConnection();
- //2、创建channel
- Channel channel = connection.createChannel();
-
- channel.basicQos(1);
-
- //3、创建队列-helloworld
- /**
- * 参数1:queue 指定队列名称
- * 参数2:durable 是否开启持久化(true)
- * 参数3:exclusive 是否排外(当前队列只能被一个消费者消费)
- * 参数4:autoDelete 如果这个队列没有其他消费者在消费,队列自动删除
- * 参数5:arguments 指定队列携带的信息
- */
- channel.queueDeclare(QUEUE_NAME, true, false, false, null);
-
- //4.开启监听Queue
- DefaultConsumer consumer = new DefaultConsumer(channel) {
- @Override
- public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
- try {
- Thread.sleep(2000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- System.out.println("小明洗澡已用水-第: " + new String(body, "UTF-8"));
-
- //手动ACK(接收信息,指定是否批量操作)
- channel.basicAck(envelope.getDeliveryTag(), false);
- }
- };
- /**
- * 参数1:queue 指定消费哪个队列
- * 参数1:deliverCallback 指定是否ACK(true:收到消息会立即告诉RabbiMQ,false:手动告诉)
- * 参数1:cancelCallback 指定消费回调
- */
- channel.basicConsume(QUEUE_NAME, false, consumer);
- System.out.println("小明开始洗澡-慢慢洗......");
-
- //5、键盘录入,让程序不结束!
- System.in.read();
-
- //6、释放资源
- channel.close();
- connection.close();
- }
-
- }
小丽洗澡:
-
- /**
- * @author rabbit
- * @version 1.0.0
- * @Description 工作模式 一个生产者,一个默认的交换机,一个队列,两个消费者
- * 需要在consumer消费者端,添加Qos能力以及更改为ACK手动即可让消费者根据自己的能力消费,不是RabbitMQ默认的平均分配了
- * @createTime 2022/07/27 19:36:00
- */
- public class XLShowerConsumer {
-
- public static final String QUEUE_NAME = "SolarWaterHeater";
-
- //消费者
- public static void main(String[] args) throws Exception {
- //1、获取连对象、
- Connection connection = RabbitCommonConfig.getConnection();
- //2、创建channel
- Channel channel = connection.createChannel();
-
- channel.basicQos(1);
-
- //3、创建队列-helloworld
- /**
- * 参数1:queue 指定队列名称
- * 参数2:durable 是否开启持久化(true)
- * 参数3:exclusive 是否排外(当前队列只能被一个消费者消费)
- * 参数4:autoDelete 如果这个队列没有其他消费者在消费,队列自动删除
- * 参数5:arguments 指定队列携带的信息
- */
- channel.queueDeclare(QUEUE_NAME, true, false, false, null);
-
- //4.开启监听Queue
- DefaultConsumer consumer = new DefaultConsumer(channel) {
- @Override
- public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
- try {
- Thread.sleep(500);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- System.out.println("小丽洗澡已用水-第: " + new String(body, "UTF-8"));
-
- //手动ACK(接收信息,指定是否批量操作)
- channel.basicAck(envelope.getDeliveryTag(), false);
- }
- };
- /**
- * 参数1:queue 指定消费哪个队列
- * 参数1:deliverCallback 指定是否ACK(true:收到消息会立即告诉RabbiMQ,false:手动告诉)
- * 参数1:cancelCallback 指定消费回调
- */
- channel.basicConsume(QUEUE_NAME, false, consumer);
- System.out.println("小丽开始洗澡-快速洗......");
-
- //5、键盘录入,让程序不结束!
- System.in.read();
-
- //6、释放资源
- channel.close();
- connection.close();
- }
-
- }
2.2.3 结果验证
生产者:
- 水龙头放水成功!1升
- 水龙头放水成功!2升
- 水龙头放水成功!3升
- 水龙头放水成功!4升
- 水龙头放水成功!5升
- 水龙头放水成功!6升
- 水龙头放水成功!7升
- 水龙头放水成功!8升
- 水龙头放水成功!9升
- 水龙头放水成功!10升
- 水龙头放水成功!11升
- 水龙头放水成功!12升
- 水龙头放水成功!13升
- 水龙头放水成功!14升
- 水龙头放水成功!15升
- 水龙头放水成功!16升
- 水龙头放水成功!17升
- 水龙头放水成功!18升
- 水龙头放水成功!19升
- 水龙头放水成功!20升
小明:
- 小明开始洗澡-慢慢洗......
- 小明洗澡已用水-第: 2升
- 小明洗澡已用水-第: 7升
- 小明洗澡已用水-第: 12升
- 小明洗澡已用水-第: 17升
小丽:
- 小丽开始洗澡-快速洗......
- 小丽洗澡已用水-第: 1升
- 小丽洗澡已用水-第: 3升
- 小丽洗澡已用水-第: 4升
- 小丽洗澡已用水-第: 5升
- 小丽洗澡已用水-第: 6升
- 小丽洗澡已用水-第: 8升
- 小丽洗澡已用水-第: 9升
- 小丽洗澡已用水-第: 10升
- 小丽洗澡已用水-第: 11升
- 小丽洗澡已用水-第: 13升
- 小丽洗澡已用水-第: 14升
- 小丽洗澡已用水-第: 15升
- 小丽洗澡已用水-第: 16升
- 小丽洗澡已用水-第: 18升
- 小丽洗澡已用水-第: 19升
- 小丽洗澡已用水-第: 20升
从结果看:实现了我们期望的结果,小丽用的水多了,小明用的水少,大家都洗好了。
此模式我们最应该注意的就是平均模式与公平模式的实现,这里是靠消费者的手工确认机制来实现的。
String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;
如上代码中的 第二个参数 autoAck,
true的时候,属于自动确认,消费者一旦接收到此消息,就会发回确认消息给Broker,Broker会从队列中删除掉此消息。当有多个消费者注册到同一个queue时,会默认轮询分发。
false的时候,属于手动确认,消费者虽然接收到了消息,还需要执行一个 方法来告诉Broker才行
channel.basicAck(envelope.getDeliveryTag(), false);
若没有告诉Broker,Broker还会将此消息再次发送。
同时此种情况下还需要使用此方法:
void basicQos(int prefetchCount) throws IOException;
指定每次消费抓取的数量
maximum number of messages that the server will deliver
这样我们通过这些改造,实现了,消费者消费的时候,消费一个告诉Broker删除一个,没消费的时候就不要给消费者投递了,最终实现了公平消费。