rabbitMQ对优先级队列的使用

注意事项：
1.队列设置优先级 权制范围（0-255）推荐0-10 否则浪费CPU与内存
2.发消息时给消息设置优先级
3.消息需要完全事先在队列中，在被消费者消费 会被排序，否则边生产边消费不会达到预期的队列优先效果。

优先级队列：
0-255越大越优先
推荐：0-10 CPU性能友好

先生产者生产消息：

package com.esint.rabbitmq.work07;

import com.esint.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

public class Producer {

    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMQUtils.getChannel();

        Map<String, Object> arguments = new HashMap<>();

        //队列设置优先即参数范围
        arguments.put("x-max-priority",10);//官方允许0-255 此处设置10 允许优先级范围 0-10  不设置过大 浪费CPU和内存

        channel.queueDeclare(QUEUE_NAME,false,false,false,arguments);

        //注意上面要求的 需要事先 发送完毕消息  才能体显现优先级的消息优化排序
        for (int i = 0; i < 10; i++) {
            String message = "msssage" + i;

            if(i == 3 ){
                //本实验的核心操作再次 在这里构建优先级设置参数 设置这个优先级的值需要在前面设置对立参数范围内
                AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().priority(5).build();
                channel.basicPublish("",QUEUE_NAME,properties,message.getBytes(StandardCharsets.UTF_8));
            }else{
                channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
            }

        }
    }
}

消费者消费消息：

package com.esint.rabbitmq.work07;

import com.esint.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;

public class Comsumer {

    public static final String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMQUtils.getChannel();

        DeliverCallback deliverCallback = ( consumerTag, message)->{
            System.out.println(new String(message.getBody(),"UTF-8"));
        };
        CancelCallback cancelCallback = (consumerTag)->{

        };
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);

    }
}

用到的工具类：

package com.esint.rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class RabbitMQUtils {

    public static Channel getChannel()throws Exception{
        //创建链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();

        //设置链接
        connectionFactory.setHost("192.168.43.37");
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");

        //链接工厂创建链接
        Connection connection = connectionFactory.newConnection();

        //获取信道
        Channel channel = connection.createChannel();

        return channel;

    }

    public static final void Sleep(int nums)  {
        try {
            Thread.sleep(nums * 1000);
        }catch (InterruptedException _ignored){
            Thread.currentThread().interrupt();
        }
    }
}

产生的结果：

Connected to the target VM, address: '127.0.0.1:59793', transport: 'socket'
msssage3
msssage0
msssage1
msssage2
msssage4
msssage5
msssage6
msssage7
msssage8
msssage9

代码007普通

打赏 收藏 海报 链接