rabbitmq优先级队列_rabbitmq主从模式

大家好，又见面了，我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全系列IDE稳定放心使用

优先级队列：此队列中的消息可以拥有优先级属性，在发送有优先级属性的消息到此队列时，优先级属性能够生效。优先级高的消息得以提早消费，消息优先级的最大值由队列的属性决定。超出队列的最大值按最大值算。

Map<String, Object> priority = new HashMap<String, Object>();
priority.put("x-max-priority", 10);
channel.queueDeclare(QUEUE_NAME, false, false, false, priority);

为了看到区别，声明了一个优先级queue2队列同时声明了一个不具优先级的队列sec_queue。

声明完队列后，发送消息并且声明消息优先级

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
// 设置TTL=6000ms
if (i>80){ 
   
    builder.priority(10);
 }
AMQP.BasicProperties properties = builder.build();
//指定发送消息到哪个路由，以及他的路由键,消息等
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, true,properties, message.getBytes());

当消息被路由到具有优先级属性的队列后，消费者再去消费就会先消费具有高优先级的消息


注意：
这个是有前提的: 只有在生产者的生产速度大于消费者，或者Broker有消息堆积的情况下才生产。如果产生一条就消费一条就没什么意义了

代码实现：

public class Send { 

//队列名
private final static String QUEUE_NAME = "queue2";
//死信队列名
private final static String SEC_QUEUE_NAME = "sec_queue";
//路由器名
private final static String EXCHANGE_NAME = "exchange";
//绑定键
private final static String BINDING_KEY = "exchange";
//路由键
private final static String ROUTING_KEY = "exchange";
private static Connection connection =null;
private static Channel channel = null;
public static void main(String[] args) { 

Map<String, Object> priority = new HashMap<String, Object>();
priority.put("x-max-priority", 10);
try{ 

// 获取到连接以及mq通道
connection = ConnectionUtil.getConnection();
// 从连接中创建通道
channel = connection.createChannel();
//声明了一个direct 类型的交换器
channel.exchangeDeclare(EXCHANGE_NAME,"direct",true,false,null);
// 声明优先级队列
channel.queueDeclare(QUEUE_NAME, false, false, false, priority);
//将路由与队列绑定，再为绑定的路径赋值一个绑定键
channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,BINDING_KEY);
//声明队列
channel.queueDeclare(SEC_QUEUE_NAME, false, false, false,null);
//将路由与队列绑定，再为绑定的路径赋值一个绑定键
channel.queueBind(SEC_QUEUE_NAME,EXCHANGE_NAME,BINDING_KEY);
//发送数据
for (int i=0;i<100;i++){ 

// 消息内容
String message = "Hello World!"+i;
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
// 设置TTL=6000ms
if (i>80){ 

builder.priority(10);
}
AMQP.BasicProperties properties = builder.build();
//指定发送消息到哪个路由，以及他的路由键,消息等
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, true,properties, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
Thread.sleep(10);
}
}catch (Exception e){ 

e.printStackTrace();
}finally { 

//关闭通道和连接
try { 

channel.close();
connection.close();
} catch (IOException e) { 

e.printStackTrace();
}
}
}
}