注意事项:
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