RabbitMQ笔记二十一：priority queue

作者: 二月_春风 | 来源:发表于2017-10-29 19:37 被阅读190次

RabbitMQ笔记二十一：priority queue
Python 3 Queue编程
C++ STL priority_queue 使用说明
TypeScript 数据结构
C++优先队列
优先队列与结构体
Priority Queue
Priority Queue
Concerning Priority Queue(With G
heap

优先级队列(priority queue)

创建具有优先级属性的队列

示列

生产端：

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.ComponentScan;

import java.util.Random;
import java.util.concurrent.TimeUnit;

@ComponentScan
public class Application {

    private static MessageProperties getmessageProperties(){
        int priority = new Random().nextInt(5);
        System.out.println("====优先级==="+priority);
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text");
        messageProperties.setPriority(priority);
        return messageProperties;

    }
    public static void main(String[] args) throws Exception{
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);

        byte[] body = "hello world".getBytes();

        //一次性发送10条消息，优先级分别是1到10
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.send("","zhihao.miao.user",new Message(body,getmessageProperties()));
        }

        TimeUnit.SECONDS.sleep(30);

        context.close();
    }
}

配置：

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MQConfig {

    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setUri("amqp://zhihao.miao:123456@192.168.1.131:5672");
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        return rabbitAdmin;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }

}

控制台打印：

====优先级===1
十月 29, 2017 2:12:52 下午 org.springframework.amqp.rabbit.connection.CachingConnectionFactory createBareConnection
信息: Created new connection: connectionFactory#1184ab05:0/SimpleConnection@6a400542 [delegate=amqp://zhihao.miao@192.168.1.131:5672/, localPort= 52105]
====优先级===1
====优先级===4
====优先级===3
====优先级===3
====优先级===1
====优先级===3
====优先级===2
====优先级===1
====优先级===4

消费端进行消费，首先看管控台，

控制台中get是按照顺序进行获取到的

代码消费呢？
应用启动类：

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.ComponentScan;

import java.util.concurrent.TimeUnit;

@ComponentScan
public class Application {
    public static void main(String[] args) throws Exception{
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);

        System.out.println(rabbitTemplate);
        TimeUnit.SECONDS.sleep(40);
        context.close();
    }
}

配置

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MQConfig {

    @Bean
    public ConnectionFactory connectionFactory(){
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setUri("amqp://zhihao.miao:123456@192.168.1.131:5672");
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        return rabbitAdmin;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }


    @Bean
    public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames("zhihao.miao.user");
        container.setDefaultRequeueRejected(false);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(new ChannelAwareMessageListener(){
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                System.out.println("=====消费消息======");
                System.out.println("消息的优先级是："+message.getMessageProperties().getPriority()+
                        " 消息内容是："+new String(message.getBody()));
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }
        });
        return container;
    }
}

信息: Created new connection: connectionFactory#687681be:0/SimpleConnection@61b4ee6c [delegate=amqp://zhihao.miao@192.168.1.131:5672/, localPort= 52155]
=====消费消息======
消息的优先级是：4消息内容是：hello world
org.springframework.amqp.rabbit.core.RabbitTemplate@7193666c
=====消费消息======
消息的优先级是：4消息内容是：hello world
=====消费消息======
消息的优先级是：3消息内容是：hello world
=====消费消息======
消息的优先级是：3消息内容是：hello world
=====消费消息======
消息的优先级是：3消息内容是：hello world
=====消费消息======
消息的优先级是：2消息内容是：hello world
=====消费消息======
消息的优先级是：1消息内容是：hello world
=====消费消息======
消息的优先级是：1消息内容是：hello world
=====消费消息======
消息的优先级是：1消息内容是：hello world
=====消费消息======
消息的优先级是：1消息内容是：hello world

很明显也是按照优先级顺序来消费的。

示列2

如果我们设置的发送消息的优先级都高于队列zhihao.miao.order设置的x-max-priority属性呢？

@ComponentScan
public class Application {

    private static MessageProperties getmessageProperties(){
        int priority = new Random().nextInt(5)+10;
        System.out.println("=====优先级==="+priority);
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text");
        messageProperties.setPriority(priority);
        return messageProperties;

    }
    public static void main(String[] args) throws Exception{
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);

        byte[] body = "hello world".getBytes();

        //一次性发送10条消息，优先级分别是1到10
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.send("","zhihao.miao.order",new Message(body,getmessageProperties()));
        }

        TimeUnit.SECONDS.sleep(30);

        context.close();
    }
}

客户端发送的消息的优先级，控制台打印出：

====优先级===13
十月 29, 2017 2:25:36 下午 org.springframework.amqp.rabbit.connection.CachingConnectionFactory createBareConnection
信息: Created new connection: connectionFactory#1184ab05:0/SimpleConnection@6a400542 [delegate=amqp://zhihao.miao@192.168.1.131:5672/, localPort= 52235]
====优先级===14
====优先级===14
====优先级===13
====优先级===11
====优先级===10
====优先级===10
====优先级===12
====优先级===14
====优先级===12

消费端代码和上面一样，执行程序，验证消费消息顺序

=====消费消息======
消息的优先级是：13 消息内容是：hello world
=====消费消息======
消息的优先级是：14 消息内容是：hello world
org.springframework.amqp.rabbit.core.RabbitTemplate@7193666c
=====消费消息======
消息的优先级是：14 消息内容是：hello world
=====消费消息======
消息的优先级是：13 消息内容是：hello world
=====消费消息======
消息的优先级是：11 消息内容是：hello world
=====消费消息======
消息的优先级是：10 消息内容是：hello world
=====消费消息======
消息的优先级是：10 消息内容是：hello world
=====消费消息======
消息的优先级是：12 消息内容是：hello world
=====消费消息======
消息的优先级是：14 消息内容是：hello world
=====消费消息======
消息的优先级是：12 消息内容是：hello world

发现没有严格的顺序，验证了如果设置的优先级大于队列设置的x-max-priority属性，则优先级失效。

发送消息之后可以通过http监控可以看到消息的详情：

http://192.168.1.131:15672/api/queues/%2F/zhihao.miao.user（/api/queues/vhost/name）

队列的信息详情

示列3

如果生产端发送很慢，消费者消息很快，则有可能不会严格的按照优先级来进行消费。

生产端每隔3s钟发送一条消息，很明显消费端消费也是按照发送的顺序。

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.ComponentScan;

import java.util.Random;
import java.util.concurrent.TimeUnit;

@ComponentScan
public class Application {

    private static MessageProperties getmessageProperties(){
        int priority = new Random().nextInt(5);
        System.out.println("====优先级==="+priority);
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType("text");
        messageProperties.setPriority(priority);
        return messageProperties;

    }
    public static void main(String[] args) throws Exception{
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class);
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);

        byte[] body = "hello world".getBytes();

        //一次性发送10条消息，优先级分别是1到10
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.send("","zhihao.miao.user",new Message(body,getmessageProperties()));
            TimeUnit.SECONDS.sleep(3);
        }

        TimeUnit.SECONDS.sleep(30);

        context.close();
    }
}

我们发现生产端生产的顺序和消费端消费的消息都是一致的。

总结：

创建优先级队列，需要增加x-max-priority参数，指定一个数字。表示最大的优先级，建议优先级设置为1～10之间。
发送消息的时候，需要设置priority属性，最好不要超过上面指定的最大的优先级。
如果生产端发送很慢，消费者消息很快，则有可能不会严格的按照优先级来进行消费。

第一，如果发送的消息的优先级属性小于设置的队列属性x-max-priority值，则按优先级的高低进行消费，数字越高则优先级越高。
第二，如果发送的消息的优先级属性都大于设置的队列属性x-max-priority值，则设置的优先级失效，按照入队列的顺序进行消费。
第三，如果消费端一直进行监听，而发送端一条条的发送消息，优先级属性也会失效。

RabbitMQ不能保证消息的严格的顺序消费。

参考资料

priority queue

RabbitMQ笔记二十一：priority queue
优先级队列(priority queue) 示列生产端：配置：控制台打印：消费端进行消费，首先看管控台， ...
Python 3 Queue编程
FIFO Queue, IFFO Queue, Priority Queue
C++ STL priority_queue 使用说明
说明优先队列std::priority_queue 可用于构造堆。比如：大顶堆：priority_queue ...
TypeScript 数据结构
1.Stack 2.Queue&priority_queue
C++优先队列
头文件< queue> priority_queue , greater > q; /...
优先队列与结构体
priority_queue q：默认将队列中的数从大到小排序；priority_queue ,greate...
Priority Queue
1、用向量实现 2、有序向量 3、列表 4、有序化列表 5、平衡搜素二叉树 6、完全二叉树 7、Complete ...
Priority Queue
介绍优先队列前我们先介绍两个基本概念：完全二叉树(Complete Binary Tree)，满二叉树(Full ...
Concerning Priority Queue(With G
Listed below is an impl of priority queue(with generics, ...
heap
最小优先：priority_queue ,greater >Q_minfirst;最大优先：priority_q...

RabbitMQ笔记二十一：priority queue

优先级队列(priority queue)

示列

示列2

示列3

总结：

参考资料

相关文章