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);
}
}
}
代码中判断i==3,i==5,i==9这三个消息的权重相等5且高于其他消息,理应提前被处理,代码运行结果为:
测试结果:
查看运行结果可以验证优先级队列的作用。
当然,在消费端速度大于生产端速度,且broker中没有消息堆积的话,对发送的消息设置优先级也没什么实际意义,因为发送端刚发送完一条消息就被消费端消费了,那么就相当于broker至多只有一条消息,那么对于单条消息来说优先级是没有什么意义的。
参考链接
1、https://www.rabbitmq/priority.html
2、http://zkread/article/1246857.html
更多推荐
Rabbitmq priority 优先级
发布评论