2025年rabbitmq优先级队列_rabbitmq主从模式

编程汇总 • 2025-03-22 22:01 • 阅读 15

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

Map priority = new HashMap();

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 priority = new HashMap();

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();

}

}

}

}

2025年rabbitmq优先级队列_rabbitmq主从模式

相关推荐