RabbitMQ03_work模型的消息生产和消费

r囧r小猫 2022-12-19 03:57 134阅读 0赞

**RabbitMQ03\_work模型的消息生产和消费**

多个消费者共同消费队列中的信息：  
![在这里插入图片描述][20201105140550551.png_pic_left]

*  **消息生产者：**

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        for(int i=1;i<=20;i++) { 
            channel.basicPublish("", "work", null, (i+"hello work queue").getBytes());
        }
        RabbitMQUtils.closeConnectionAndChannel(channel, connection);
    }

*  **消息消费者1：**

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        //自动确认设为true
        channel.basicConsume("work", true,new DefaultConsumer(channel) { 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException { 
                System.out.println("消费者1--"+new String(body));
            }
        });
    }

*  **消息消费者2：**

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        //自动确认设为true
        channel.basicConsume("work", true,new DefaultConsumer(channel) { 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException { 
                System.out.println("消费者2--"+new String(body));
            }
        });
    }

消费者1--2hello work queue
    消费者1--4hello work queue
    消费者1--6hello work queue
    消费者1--8hello work queue
    消费者1--10hello work queue
    消费者1--12hello work queue
    消费者1--14hello work queue
    消费者1--16hello work queue
    消费者1--18hello work queue
    消费者1--20hello work queue

消费者2--1hello work queue
    消费者2--3hello work queue
    消费者2--5hello work queue
    消费者2--7hello work queue
    消费者2--9hello work queue
    消费者2--11hello work queue
    消费者2--13hello work queue
    消费者2--15hello work queue
    消费者2--17hello work queue
    消费者2--19hello work queue

**可以观察发现，它默认是一种轮询的平均分配的效果**

*  修改消费者1,使其消费速度变慢：

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        //自动确认设为true
        channel.basicConsume("work", true,new DefaultConsumer(channel) { 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException { 
                System.out.println("消费者1--"+new String(body));
                try { 
                    Thread.sleep(2000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        });
    }

**再次测试，发现消息仍然是平均分配的，但消费者2很快完成，消费者1却花了20s才完成**

*  修改消费者1，使其出错无法执行消费：

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        //自动确认设为true
        channel.basicConsume("work", true,new DefaultConsumer(channel) { 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException { 
                int i = 1/0;
                System.out.println("消费者1--"+new String(body));
                try { 
                    Thread.sleep(2000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        });
    }

结果发现，消费者2消费了10条消息，但消费者1出错，1条消息都没有消费  
从管理界面中发现，待处理消息为0，未确认消息为0，说明丢失了10条消息！！  
![在这里插入图片描述][20201105144529662.png_pic_left]

*  修改消费者1和消费者2，关闭自动确认机制，选择手动确认，并设定每次只允许消费一条消息：

消费者1：

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //设定每次只能从队列中取1个消息来消费
        channel.basicQos(1);
        channel.queueDeclare("work",true,false,false,null);
        //自动确认设为false
        channel.basicConsume("work", false,new DefaultConsumer(channel) { 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException { 
                int i = 1/0;
                System.out.println("消费者1--"+new String(body));
                try { 
                    Thread.sleep(2000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
                //手动确认消息，参数1为消息的标识，参数2表示每次只允许确认1个消息
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }

消费者2：

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        //设定每次只能从队列中取1个消息来消费
        channel.basicQos(1);
        channel.queueDeclare("work",true,false,false,null);
        //自动确认设为false
        channel.basicConsume("work", false,new DefaultConsumer(channel) { 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException { 
                System.out.println("消费者2--"+new String(body));
                //手动确认消息，参数1为消息的标识，参数2表示每次只允许确认1个消息
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }

结果发现，消费者1出错无法执行，消费者2消费了所有消息，实现了能者多劳，且无消息丢失：

消费者2--2hello work queue
    消费者2--1hello work queue
    消费者2--3hello work queue
    消费者2--4hello work queue
    消费者2--5hello work queue
    消费者2--6hello work queue
    消费者2--7hello work queue
    消费者2--8hello work queue
    消费者2--9hello work queue
    消费者2--10hello work queue
    消费者2--11hello work queue
    消费者2--12hello work queue
    消费者2--13hello work queue
    消费者2--14hello work queue
    消费者2--15hello work queue
    消费者2--16hello work queue
    消费者2--17hello work queue
    消费者2--18hello work queue
    消费者2--19hello work queue
    消费者2--20hello work queue

[20201105140550551.png_pic_left]: /images/20221120/c9708fbbc62641eca9a32fa914179766.png
[20201105144529662.png_pic_left]: https://img-blog.csdnimg.cn/20201105144529662.png#pic_left