RabbitMQ笔记二十一：priority queue

优先级队列(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

最后编辑于：2017.12.11 05:12:56

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 211,290评论 6赞 491
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 90,107评论 2赞 385
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 156,872评论 0赞 347
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 56,415评论 1赞 283
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 65,453评论 6赞 385
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 49,784评论 1赞 290
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,927评论 3赞 406
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 37,691评论 0赞 266
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 44,137评论 1赞 303
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 36,472评论 2赞 326
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,622评论 1赞 340
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 34,289评论 4赞 329
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,887评论 3赞 312
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,741评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,977评论 1赞 265
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 46,316评论 2赞 360
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 43,490评论 2赞 348

RabbitMQ笔记二十一 ：priority queue

优先级队列(priority queue)

示列

示列2

示列3

总结：

参考资料

推荐阅读更多精彩内容

RabbitMQ笔记二十一：priority queue