rabbitmq 优先级队列

前言

首先什么是优先级队列呢？优先级队列是指优先级高的消息往往放在队列的head头部，相比低优先级的消息，要优先投递给消费者进行处理。接下来，将介绍在rabbitmq中如何实现优先级队列。

具体使用

版本要求：rabbitmq 3.5.0以上

1. 定义优先级队列

   Channel ch = ...;
   Map<String, Object> args = new HashMap<String, Object>();
   args.put("x-max-priority", 10); // 定义队列的最大优先级值
   ch.queueDeclare("my-priority-queue", true, false, false, args);

   或者通过后台管理界面定义
   

2. 发送优先级消息

   消息的优先级取值范围是无符号byte[0, 255]，当消息没有设置优先级时，则以最低优先级0对待；当消息的优先级大于queue中设置的最大优先级时，则以queue中设置的最大优先级对待；对于相同优先级的消息，先进的排在前面，后进的排在后面。

   AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
   builder.priority(5); //设置消息优先级为5
   AMQP.BasicProperties properties = builder.build();
   channel.basicPublish("my-priority-exchange","my-priority-queue-key", properties, "nop".getBytes());

示例代码

public static void main(String[] args) throws IOException {
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("127.0.0.1");
    connectionFactory.setPort(5672);
    connectionFactory.setUsername("test");
    connectionFactory.setPassword("1234");
    connectionFactory.setVirtualHost("/test");
    connectionFactory.setAutomaticRecoveryEnabled(true);
    Connection conn = connectionFactory.newConnection();
    final Channel channel = conn.createChannel();

    // 创建消费者
    channel.basicQos(1);
    final CountDownLatch latch = new CountDownLatch(1);
    channel.basicConsume("my-priority-queue", new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
            byte[] body) throws IOException {
            System.out.println(new String(body, "utf-8"));
            try {
               // 等待下面的生产者将所有消息投递到queue，在返回给第一个消息确认回复
                latch.await();
            } catch (InterruptedException e){} 
            channel.basicAck(envelope.getDeliveryTag(), false); // 确认消息已收到
        }
    });

    // 投递优先级消息
    String msg = "msg-priority-";
    AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
    for (int i = 0; i < 20;)
    {
        builder.priority(i); // 设置消息优先级为i
        AMQP.BasicProperties properties = builder.build();
        channel.basicPublish("my-priority-exchange", "my-priority-queue-key", properties, (msg + i).getBytes());
        i = i + 2;
    }
    // 通知消费者继续消费queue中的消息
    latch.countDown();
}

注：要看到优先级队列的效果，必须要保证queue中同时存在不同优先级的消息且消费者处于忙碌状态。

应用场景

1. 各业务系统投递给短信平台的短信，通常是由优先级之分的，比如短信验证码这一类的就要求及时性，需要短信平台那边优先处理 ——属于紧急重要；有些通知类的短信就不需要那么高的及时性，可以稍后处理等。
2. ……

参考链接

1. http://www.rabbitmq.com/priority.html
2. http://www.rabbitmq.com/community-plugins.html