快捷導航

rabbitmq結(jié)合spring實現(xiàn)消息隊列優(yōu)先級的方法

更新時間：2017年02月16日 15:28:45 作者：安南0558

本篇文章主要介紹了rabbitmq結(jié)合spring實現(xiàn)消息隊列優(yōu)先級的方法，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧

1.1項目背景：做一個災(zāi)情預警的消息平臺，災(zāi)情檢查系統(tǒng)需要向消息平臺里面推送消息，這里是典型的異構(gòu)系統(tǒng)的消息傳遞，我們需要選擇一個中間件作為消息隊列，調(diào)研分析了rabbitmq，zeromq，activemq，kafka等消息中間件，綜合性能，安全，可持久化等角度果斷選擇了rabbitmq作為我們的消息中間件 (其實這里是因為rabbitmq 是spring官方支持的，開發(fā)起來方便)。需求上我們有多種類型的消息，這里有緊急推送的和一般的等區(qū)分，高并發(fā)時，就會有對消息進行優(yōu)先推送的情況出現(xiàn)，于是rabbitmq消息隊優(yōu)先級的推送功能是我們需要解決的首個技術(shù)點.

1.2技術(shù)調(diào)研:這里一個概念需要說明,為什么說是消息隊列的優(yōu)先級而不是消息的優(yōu)先級,來看下消息隊列的工作原理

生產(chǎn)者生成消息打到交換機里面(如果沒有聲明交換機，會打到default exchange里面)，交換機綁定一個或多個隊列，消息進入隊列里面，消費者一直在監(jiān)聽隊列，發(fā)現(xiàn)隊列里面有消息就開始消費，這里就是一個消息傳遞的過程，queue是一個棧隊列，棧是先進先出的，就是說消息來了依次排隊，一個隊列并不能實現(xiàn)消息的插隊和優(yōu)先推送的功能。但是如果說我們的多個隊列有不同的優(yōu)先級，不同優(yōu)先級的消息通過roatingkey進入不同的隊列，優(yōu)先級高的隊列消息被優(yōu)先消費，這樣也能形成一個相對意義上的優(yōu)先級，所以說這里不是消息的優(yōu)先級而是隊列的優(yōu)先級.

1.2.1 為什么說是相對意義上的優(yōu)先級

有并發(fā)才有優(yōu)先級，如果每個消息都能被瞬間處理也不會有消息優(yōu)先推送的需求，那我們看看消息會在哪里阻塞

1，queue，很明顯高并發(fā)的時候隊列里面是會存在很多消息的，2，eschange ，高并發(fā)的時候producer發(fā)送給exchange的時候也會產(chǎn)生阻塞。

第一種情況由于我們隊列已經(jīng)定義優(yōu)先級了，所以進入隊列的消息都是同種優(yōu)先級別的，并不需要插隊。而對于第二種情況，消息在exchange時阻塞時并不能實現(xiàn)消息優(yōu)先進入隊列，依然是一個依次處理的情景，但是由于exchang到queue的處理速度極快，所有我們忽略了這塊的優(yōu)先級。

1.2.3 代碼實現(xiàn)

在rabbitmq3.5版本之前，官方并沒有實現(xiàn)隊列優(yōu)先級的功能，但論壇里面有一些插件可以實現(xiàn)(末尾附鏈接),這里我們主要說3.5版本之后的實現(xiàn)

1.2.3.1 Java代碼

Connectionconn =RabbitMQConnectionUtil.getRabbitmqConnection();//創(chuàng)建連接 
    Channelchannel = conn.createChannel();//創(chuàng)建channel 
    Map<String,Object> arg = newHashMap<String, Object>();  
    arg.put("x-max-priority",10); //隊列的屬性參數(shù) 有10個優(yōu)先級別 
    // 聲明（創(chuàng)建）隊列 
    //channel.queueDeclare(QUEUE_NAME, false, false, false, null); 
    channel.queueDeclare(QUEUE_NAME,true,false, false, arg); 
    // 消息內(nèi)容 
    String message ="Hello World!"; 
    channel.basicPublish("",QUEUE_NAME, null, message.getBytes()); 
    BasicPropertiesprop =new BasicProperties(null, null, null, null, 1, 
    null, null, null, null, null, null, null, null,null);//消息的參數(shù)，聲明該消息的優(yōu)先級是1 
    channel.basicPublish("",QUEUE_NAME, prop, message.getBytes()); //消息發(fā)布 
    System.out.println("[x] Sent '" + message + "'"); 
    //關(guān)閉通道和連接 
    channel.close(); 
    conn.close();

客戶端看下結(jié)果:

1.2.3.2結(jié)合spring實現(xiàn)：

1.2.3.2.1 xml配置：

<beans xmlns="http://www.springframework.org/schema/beans" 
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
  xmlns:rabbit="http://www.springframework.org/schema/rabbit"  
  xsi:schemaLocation="http://www.springframework.org/schema/beans  
  http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
  http://www.springframework.org/schema/rabbit 
  http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd" > 
  <description>rabbitmq 連接服務(wù)配置</description> 
  <!-- 連接配置 --> 
  <rabbit:connection-factory id="connectionFactory" host="${rabbit.ip}" username="${rabbit.username}"  
  password="${rabbit.password}" port="${rabbit.port}" virtual-host="${rabbit.vhost}"/> 
  <rabbit:admin connection-factory="connectionFactory"/> 
  <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" /> 
  <!-- spring template聲明--> 
  <!-- 聲明一個隊列 --> 
  <rabbit:queue id="test_queue_key" name="test_queue_key" durable="true" auto-delete="false" exclusive="false"> 
   <rabbit:queue-arguments> 
    <entry key="x-max-priority"> 
      <value type="java.lang.Integer">10</value>//這個地方一定是integer的，別的不好使??！ 
    </entry> 
   </rabbit:queue-arguments> 
  </rabbit:queue> 
  <!-- 監(jiān)聽配置queues：監(jiān)聽的隊列，多個的話用逗號（,）分隔 ref：監(jiān)聽器--> 
  <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto"> 
    <rabbit:listener queue-names="test_queue_key" ref="queueListenter" method="onMessage"/> 
  </rabbit:listener-container> 
<bean id="queueListenter" class="com.DF.spring.springAMQP.QueueListener" />

1.2.3.2.2代碼部分:

producter:

AbstractApplicationContext ctx = new  
ClassPathXmlApplicationContext("classpath:/spring/rabbitmq-contextDemo2.xml"); 
RabbitTemplate amqpTemplate = ctx.getBean(RabbitTemplate.class); 
 Random random = new Random(); 
  for (int i=0; i< 1000; i++){ 
    final int priority = random.nextInt(10 - 1 + 1) + 1;//隨機的優(yōu)先級 
    amqpTemplate.convertAndSend("test_queue_key", (Object)("hello world"), new MessagePostProcessor() { 
     @Override 
     public Message postProcessMessage(Message message) throws AmqpException { 
      message.getMessageProperties().setPriority(priority); 
      return message; 
     } 
    }); 
  }

customer：

public class QueueListener implements MessageListener{ 
 @Override 
 public void onMessage(Message message) { 
  try{ 
   System.out.print("[x] 接收到的消息："+new String(message.getBody(),"utf-8")+"&&&"+"優(yōu)先級"+message.getMessageProperties().getPrority()); 
   Thread.sleep(1000); 
  }catch(Exception e){ 
   e.printStackTrace(); 
  } 
 } 
}

從客戶端看下隊列里面的消息：