Rabbitmq priority 优先级

编程入门行业动态更新时间:2024-10-27 03:40:25

Rabbitmq priority 优先级

顾名思义，具有更高优先级的队列具有较高的优先权，优先级高的消息具备优先被消费的特权。
RabbitMQ has priority queue implementation in the core as of version 3.5.0.

在系统应用中会根据业务的优先级来决定哪些内容优先被解决，那么在RabbitMQ 3.5+版本中支持了队列优先级和消息优先级，那么这里就尝试下该功能。

在rabbitmq webUI界面中，add queue是通过添加Arguments或者是在代码中添加效果是一样的,取值范围在0~255之间：

The message priority field is defined as an unsigned byte, so in practice priorities should be between 0 and 255.

具体代码如下:
生产者：

public class NewTask {


    private final static String hostname = "192.168.1.137";

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

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建连接和频道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(hostname);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        try {
            Map<String,Object> param = new HashMap<String,Object>();
            param.put("x-max-priority", 10);
            channel.queueDeclare(QUEUE_NAME, false, false, true, param);

            //send message with priority
            for(int i=0;i<10;i++) {
                AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
                if(i==9 || i==3 || i==5)
                    builder.priority(5);
                AMQP.BasicProperties properties = builder.build();
                channel.basicPublish("",QUEUE_NAME,properties,("messages-"+i).getBytes());
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            channel.close();
            connection.close();
        }
    }
}

消费者：

public class Worker1 {

    private final static String hostname = "192.168.1.137";

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

    public static void main(String[] argv) throws java.io.IOException,
            java.lang.InterruptedException, TimeoutException {

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(hostname);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, false, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.out.println(msg);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        }
    }

}