当前位置: 首页>后端>正文

多个rabbitmq容器 rabbitmq可以创建多少个队列

1.介绍

简单队列有个缺点,简单队列是一一对应的关系,即点对点,一个生产者对应一个消费者,按照这个逻辑,如果我们有一些比较耗时的任务,也就意味着需要大量的时间才能处理完毕,显然简单队列模式并不能满足我们的工作需求,我们今天再来看看工作队列。

多个rabbitmq容器 rabbitmq可以创建多少个队列,多个rabbitmq容器 rabbitmq可以创建多少个队列_rabbitmq,第1张

一个生产者对应多个消费者,但是只能有一个消费者获得消息!!!

创建maven项目,导入依赖

<dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.7.3</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.25</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>

2.消息生产者

发送50条消息

/**
 * 消息生产者
 * @Author liuqingsen
 * @Date 2020/6/21 15:10
 */
public class Producer {

    //队列名
    private static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();
        //创建通道
        Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);

        for ( int i = 0 ; i < 50 ; i++){
            String msg = "生产者消息=>"+i;
            System.out.println("生产者发送消息:" + msg);
            //发送消息
            channel.basicPublish("",QUEUE_NAME,null,msg.getBytes()  );
            //模拟发送消息延时,便于演示多个消费者竞争接受消息
            Thread.sleep(i*10);
        }

        channel.close();
        connection.close();
    }
}

3.消息消费者

消费者1,等待10毫秒,模拟处理详细效率

/**
 * 消费者1
 * @Author liuqingsen
 * @Date 2020/6/21 15:19
 */
public class Customer_1 {

    //队列名
    private static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //定义一个消费者
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            //一旦该队列有消息就会触发
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body,"utf-8");
                System.out.println("消费者[1]获取消息:" + msg);

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        //监听队列
        channel.basicConsume(QUEUE_NAME,false,defaultConsumer);
    }
}

消费者2,等待时间3000毫秒,模拟处理消息效率

/**
 * 消费者2
 * @Author liuqingsen
 * @Date 2020/6/21 15:39
 */
public class Customer_2 {

    //队列名
    private static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //定义一个消费者
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            //一旦该队列有消息就会触发
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body,"utf-8");
                System.out.println("消费者[2]获取消息:" + msg);

                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        //监听队列
        channel.basicConsume(QUEUE_NAME,false,defaultConsumer);
    }
}

生产者控制台打印:

多个rabbitmq容器 rabbitmq可以创建多少个队列,多个rabbitmq容器 rabbitmq可以创建多少个队列_java_02,第2张

消费者1控制台打印:

多个rabbitmq容器 rabbitmq可以创建多少个队列,多个rabbitmq容器 rabbitmq可以创建多少个队列_rabbitmq_03,第3张

消费者2控制台打印:

多个rabbitmq容器 rabbitmq可以创建多少个队列,多个rabbitmq容器 rabbitmq可以创建多少个队列_java_04,第4张

可以看出是轮询分发消息的

存在的问题:消费者处理能力没有分配好,我把消息分发给你们,你们什么时候处理完,不管我事!
消费者1比消费者2执行效率更高,分配了相同数量的消息,这样显然是不合理的!

4.能者多劳

其实发生上述问题的原因是 RabbitMQ 收到消息后就立即分发出去,而没有确认各个工作者未返回确认的消息数量,类似于TCP/UDP中的UDP,面向无连接。

因此我们可以使用 basicQos 方法,并将参数 prefetchCount 设为1,告诉 RabbitMQ 我每次值处理一条消息,你要等我处理完了再分给我下一个。这样 RabbitMQ 就不会轮流分发了,而是寻找空闲的工作者进行分发。

消费者2就改为:
1. 最大分发次数
2. 手动回执消息(注意:MQ没有收到回执,代表没有执行结束,则只会发送初始分发数量的消息!)

/**
 * 消费者2
 * @Author liuqingsen
 * @Date 2020/6/21 15:39
 */
public class Customer_2 {

    //队列名
    private static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //保证一次只分发一次 限制发送给同一个消费者 不得超过一条消息
        channel.basicQos(1);
        //定义一个消费者
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            //一旦该队列有消息就会触发
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body,"utf-8");
                System.out.println("消费者[2]获取消息:" + msg);

                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    //手动回执消息,默认自动
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        //监听队列
        channel.basicConsume(QUEUE_NAME,false,defaultConsumer);
    }
}

消费者1控制台打印

多个rabbitmq容器 rabbitmq可以创建多少个队列,多个rabbitmq容器 rabbitmq可以创建多少个队列_rabbitmq_05,第5张

消费者2控制台打印

多个rabbitmq容器 rabbitmq可以创建多少个队列,多个rabbitmq容器 rabbitmq可以创建多少个队列_java_06,第6张

可以明显看出,执行效率高的(消费者1)分配的消息更多,执行效率低的分配的消息更少。

5.消息持久化

当有多个消费者同时收取消息,且每个消费者在接收消息的同时,还要处理其它的事情,且会消耗很长的时间。在此过程中可能会出现一些意外,比如消息接收到一半的时候,一个消费者死掉了。

这种情况要使用消息接收确认机制,可以执行上次宕机的消费者没有完成的事情。

但是在默认情况下,我们程序创建的消息队列以及存放在队列里面的消息,都是非持久化的。当RabbitMQ死掉了或者重启了,上次创建的队列、消息都不会保存。

参数配置:
参数配置一:生产者创建队列声明时,修改第二个参数为 true

/**3.创建队列声明 */
channel.queueDeclare(QUEUE_NAME, true, false, false, null);

参数配置二:生产者发送消息时,修改第三个参数为MessageProperties.PERSISTENT_TEXT_PLAIN

for (int i = 1; i <= 50; i++) {
    String msg = "生产者消息_" + i;
    System.out.println("生产者发送消息:" + msg);
    /**4.发送消息 */
    channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
}



https://www.xamrdz.com/backend/3pr1928844.html

相关文章: