多消费者的情况

迷南。 2022-06-11 06:28 125阅读 0赞

模型:

![工作队列][2016-02-18_56c53cbccaaab.jpg]

生产者将5条消息发送到队列中，然后c1和c2把队列中的消息取出打印。

public class Producer {
            private static final String NAME="multiConsumer";
        
            public static void main(String[] args) throws IOException {
                ConnectionFactory factory=new ConnectionFactory();
                factory.setHost("127.0.0.1");
                Connection connection=factory.newConnection();
                Channel channel=connection.createChannel();
                channel.queueDeclare(NAME, true, false, false, null);
                for(int i = 0 ; i < 5; i++){
                    String message = "我是生产者! " + i;
                    channel.basicPublish("", NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
                    System.out.println(" [x] Sent '" + message + "'");
                }
                channel.close();
                connection.close();
            }
        }

消费者1:

public class Consumer1 {
        private static final String NAME="multiConsumer";
        private static void doWork(String task) {
            try {
                Thread.sleep(1000); // 暂停1秒钟
            } catch (InterruptedException _ignored) {
                Thread.currentThread().interrupt();
            }
        }
        public static void main(String[] args) {
            ConnectionFactory connectionFactory=new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            try {
                Connection connection=connectionFactory.newConnection();
                final Channel channel=connection.createChannel();
                channel.queueDeclare(NAME,true,false,false,null);
    
                channel.basicQos(1);//每次取一个
                Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        String message = new String(body, "UTF-8");
                        System.out.println("Consumer1 [x] Received '" + message + "'");
                        try {
                            doWork(message);
                        } finally {
                            System.out.println("Consumer1 [x] Done");
                            // 消息处理完成确认
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        }
                    }
                };
                //消息消费完成确认
                channel.basicConsume(NAME, false, consumer);
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
    }

消费者2:

public class Consumer2 {
        private static final String NAME="multiConsumer";
        private static void doWork(String task) {
            try {
                Thread.sleep(1000); // 暂停1秒钟
            } catch (InterruptedException _ignored) {
                Thread.currentThread().interrupt();
            }
        }
        public static void main(String[] args) {
            ConnectionFactory connectionFactory=new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            try {
                Connection connection=connectionFactory.newConnection();
                final Channel channel=connection.createChannel();
                channel.queueDeclare(NAME,true,false,false,null);
    
                channel.basicQos(1);//每次取一个
    
                 Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        String message = new String(body, "UTF-8");
    
                        System.out.println("Worker2 [x] Received '" + message + "'");
                        try {
                            doWork(message);
                        } finally {
                            System.out.println("Worker2 [x] Done");
                            // 消息处理完成确认
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        }
                    }
                };
                // 消息消费完成确认
                channel.basicConsume(NAME, false, consumer);
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
    }

当生产者将消息发送给多个消费者的时候，默认是随机发送的，有可能一个消费者一直在"干活"但是另外一个一直处于空闲的状态，为此我们可以通过设置

channel.basicQos(1);

来设置消费者只有处理完当前的消息并且返回确认消息才会接收到下一个消息。

这里必须要说一下RabbitMQ的消息确认机制，消费者完成处理之后，会向RabbitMQ发送一个反馈，只有收到这个反馈，队列中进行删除这个消息，好处是如果RabbitMQ没崩消息就不会丢失，坏处是如果消费者忘记返回处理完成的消息，那么队列可能会发生内存爆满的情况。

假如RabbitMQ服务器崩掉了，那么可以通过消息的持久化来解决消息丢失的问题，可以通过下面的代码来设置。

boolean durable = true;  
channel.queueDeclare("hello", durable, false, false, null);

[2016-02-18_56c53cbccaaab.jpg]: /images/20220611/f605273166234e8a88154fe161988980.png