1、maven依賴

<!-- rabbitMQ依賴 -->
<!-- spring相關的其他依賴請參考原始碼，此處不做過多描述 -->
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>1.4.6.RELEASE</version>
</dependency>

2.1、配置rabbitMQ連線

（1）採用<rabbit/>方式配置連線

<rabbit:connection-factory id="connectionFactory" 
host="${rabbitmq.host}" 
port="${rabbitmq.port}" 
        username="${rabbitmq.username}" 
        password="${rabbitmq.password}" 
        virtual-host=""
        publisher-confirms="true"
        publisher-returns="true"
        channel-cache-size="5"
        />
 

（2）採用普通方式配置連線

<bean id="connectionFactory"
	class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
		<constructor-arg value="${rabbitmq.host}" />
		<property name="username" value="${rabbitmq.username}" />
		<property name="password" value="${rabbitmq.password}" />
		<property name="port" value="${rabbitmq.port}" />
		<!-- 快取中要維護的通道數 -->
		<property name="channelCacheSize" value="5" />
		<!-- 開啟發送確認機制 -->
		<property name="publisherConfirms" value="true"/>
		<!-- 開啟結果返回機制 -->
		<property name="publisherReturns" value="true"/>
	</bean>
 

2.2、配置RabbitAdmin

<rabbit:admin connection-factory="connectionFactory" />

2.3、定義Queue

<rabbit:queue name="queueTest" durable="true" auto-delete="false"
exclusive="false" />

常用引數解釋：

durable：是否持久化

auto-delete：是否當沒有連線時自動刪除

exclusive：是否只能由建立者使用

2.4、定義Exchange，並繫結Queue

（1）定義direct型別

<rabbit:direct-exchange name="exchangeTest"
		durable="true" auto-delete="false">
	<rabbit:bindings>
<!-- 此處沒有指定RoutingKey -->
			<rabbit:binding queue="queueTest" key=""></rabbit:binding>
		</rabbit:bindings>
</rabbit:direct-exchange>

（2）定義topic型別

<rabbit:topic-exchange name="topicexchangetest" durable="true" auto-delete="false">
		<rabbit:bindings>
			<rabbit:binding queue="" key=""></rabbit:binding>
		</rabbit:bindings>
</rabbit:topic-exchange>

（3）定義fanout型別

<rabbit:fanout-exchange name=""></rabbit:topic-exchange>

（4）定義headers型別

<rabbit:headers-exchange name=""></rabbit:topic-exchange>

常用引數解釋：

durable：是否持久化

auto-delete：是否當沒有連線時自動刪除

注意：在此例中，如果不把Exchange和queue進行繫結，傳送訊息的時候ConfirmCallback依然正常執行，因為訊息到達了Exchange。但是ReturnCallback就會執行回撥方法，傳回錯誤資訊：NO_ROUTE。Exchange沒有找到指定的Queue，丟棄這條訊息並把訊息返回給生產者。回撥順序是ReturnCallback在前，ConfirmCallback在後。

2.5、定義rabbit template

（1）普通方式配置rabbitTemplate

<bean class="org.springframework.amqp.rabbit.core.RabbitTemplate" id="rabbitTemplate">
		<property name="connectionFactory" ref="connectionFactory" />
		<property name="exchange" value="exchangeTest" />
		
		<!-- 啟動AMQP協議層面事務機制來解決傳送確認機制，但是採用事務機制實現會降低RabbitMQ的訊息吞吐量。
		RabbitMQ團隊為我們拿出了更好的方案，即採用傳送方確認模式，事務機制和confirmCallback只能二選一 -->
		<!-- <property name="channelTransacted" value="true" /> -->

		<!-- mandatory 監聽是否有符合的佇列 -->
		<property name="mandatory" value="true"/>
		
		<!-- 設定傳送確認回執監聽方法 -->
		<property name="confirmCallback" ref="confirmcallback" /> 
		<!-- 設定結果返回監聽方法 -->
		<property name="returnCallback" ref="MyReturnCallback"/>
		<!-- 設定訊息轉換 -->
		<property name="messageConverter" ref="JsonMessageConverter" />
	</bean>

（2）<rabbit/>方式配置rabbitTemplate和相關監聽器

<rabbit:template id="rabbitTemplate" connection-factory="connectionFactory" 
		exchange="exchangeTest" 
		mandatory="true" 
		confirm-callback="confirmcallback"
		return-callback="MyReturnCallback" 
		encoding="UTF-8"
		message-converter="JsonMessageConverter"
		>
	</rabbit:template>

常用引數解釋：

mandatory：監聽是否有符合的佇列

confirm-callback：訊息確認回撥類

return-callback：mandatory監聽結果回撥類

encoding：編碼

message-converter：訊息轉化類

（1）其他相關配置

<bean id="JsonMessageConverter" class="org.springframework.amqp.support.converter.JsonMessageConverter">
    <!-- if necessary, override the DefaultClassMapper -->
    <property name="classMapper" ref="customClassmapper" />
</bean>

<bean id="customClassmapper" class="pruducer.com.taikang.tkh.message.common.CustomClassMapper"/>

<bean id="confirmcallback" class="pruducer.com.taikang.tkh.message.common.ConfirmReturnBack"/>

<bean id="MyReturnCallback" class="pruducer.com.taikang.tkh.message.common.MyReturnCallback"/>

2.6配置Consumer

<rabbit:listener-container  
	prefetch="10" 
	connection-factory="connectionFactory"  
	acknowledge="manual">
	<rabbit:listener  queues="queueTest" ref="messageReceiver" />
</rabbit:listener-container>

<!-- 自定義訊息接收者 -->
<bean id="messageReceiver" class="consumer.com.taikang.tkh.message.listener.MessageConsumer"></bean>

常用引數解釋：

prefetch：訊息預取數目為每次接收10條。

acknowledge：acknowledge="manual":意為表示該消費者的ack方式為手動 ；acknowledge="auto"表示自動。

3、spring-rabbit常用方法介紹

3.1、訊息傳送

/**
* 傳送訊息到預設的交換機和佇列（不帶有自定義引數）
* @param messageObject 訊息物件
* @return boolean 傳送標記 
*/
RabbitTemplate.convertAndSend(messageObject);
 
/**
* 傳送訊息到預設的交換機和佇列
* @param messageObject 訊息物件
* @param messageObject 自定義引數，在監聽器ConfirmCallback中可以取到。
* @return boolean 傳送標記 
*/
RabbitTemplate.correlationConvertAndSend(messageObject,correlationdata);
 
/**
* 傳送訊息到指定的佇列
* @param queue           佇列名稱
* @param messageObject   訊息物件
* @param messageObject 自定義引數，在監聽器ConfirmCallback中可以取到。
* @return boolean 傳送標記 
*/
RabbitTemplate.convertAndSend(queue, messageObject,correlationdata);
 
/**
* 傳送訊息到指定的交換機和佇列
* @param exchange       交換機名稱
* @param queue          佇列名稱
* @param messageObject 自定義引數，在監聽器ConfirmCallback中可以取到。
* @return boolean 傳送標記 
*/
RabbitTemplate.convertAndSend(exchange,queue,messageObject,correlationdata);
 
/**
* 傳送訊息到預設的交換機和佇列（不帶有自定義引數）
Send方法還有很多，此處只列舉一種
* @param Message AMQP封裝的訊息物件
* @return void 
*/
RabbitTemplate.send(Message message);

沒有convert的需要自己將傳送的物件轉換為Message物件。

3.2、Confirm監聽

public class ConfirmReturnBack implements ConfirmCallback{
	/**
	 * Confirmation callback.
	 * @param correlationData 回撥的相關資料.
	 * @param ack true for ack, false for nack
	 * @param cause 專門給NACK準備的一個可選的原因，其他情況為null。
	 */
	public void confirm(CorrelationData correlationdata, boolean ack, String cause) {
		System.out.println("Exchange接收是否成功（ack）: " + ack + "。 返回的使用者引數（correlationData）: " + correlationdata + "。NACK原因（cause） : " + cause);
	}
}

注意：CorrelationData 是在傳送訊息時傳入回撥方法的引數，可以用於區分訊息物件。 CorrelationData物件中只有一個屬性 String id。通過這個引數，我們可以區分當前是傳送哪一條訊息時的回撥，並通過ack引數來進行失敗重發功能。

3.3、Return監聽

/**
 * 實現此方法在basicpublish失敗時回撥
 * 相當於 ReturnListener的功能。
 * 在釋出訊息時設定mandatory等於true，
 * 監聽訊息是否有相匹配的佇列，沒有時ReturnCallback將執行returnedMessage方法，訊息將返給傳送者 
 */
public class MyReturnCallback implements ReturnCallback {

	public void returnedMessage(Message message, int replyCode, String replyText, 
			String exchange, String routingKey) {
		try {
			System.out.println("訊息傳送進指定佇列失敗：失敗原因（+replyText）："+replyText
					+";錯誤程式碼（replyCode）："+replyCode
					+";訊息物件："+new String(message.getBody(),"UTF-8")
					+"exchange:"+exchange
					+"routingKey:"+routingKey);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
	}
}

3.4、訊息處理

/**
 * 訊息到達消費者監聽類
 */
public class MessageConsumer implements ChannelAwareMessageListener {
	/**
	 * 處理收到的rabbit訊息的回撥方法。
	 * @param message AMQP封裝訊息物件
	 * @param channel 通道物件，可以進行確認回覆
	 * @throws Exception Any.
	 */
	public void onMessage(Message message, Channel channel) throws Exception {
		try {
			String srt2=new String(message.getBody(),"UTF-8");
			System.out.println("消費者收到訊息："+srt2);
			//成功應答
			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
		} catch (Exception e) {
			e.printStackTrace();
			//失敗應答
			channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
		}
	}
}

注意：ChannelAwareMessageListener和MessageListener兩個介面都實現對訊息到達消費者時的監聽，只不過MessageListener的onMessage方法沒有Channel引數，不能實現訊息手動應答功能。

4、Demo專案完整程式碼

【四川樂山程式設計師聯盟，歡迎大家加群相互交流學習5 7 1 8 1 4 7 4 3】