轉載於：https://blog.csdn.net/LeiXiaoTao_Java/article/details/78924863
1、maven依賴
-
<dependency>
-
<groupId>commons-lang</groupId>
-
<artifactId>commons-lang</artifactId>
-
<version>
2.3</version>
-
</dependency>
-
-
<dependency>
-
<groupId>com.rabbitmq</groupId>
-
<artifactId>amqp-client</artifactId>
-
<version>
3.4.1</version>
-
</dependency>
2、RabbitMQ重要方法介紹（基本常用的）
2.1、創建連接
-
// 創建連接工廠
-
ConnectionFactory cf =
new ConnectionFactory();
-
// 設置rabbitmq服務器IP地址
-
cf.setHost(
"*.*.*.*");
-
// 設置rabbitmq服務器用戶名
-
cf.setUsername(
"***");
-
// 設置rabbitmq服務器密碼
-
cf.setPassword(
"***");
-
// 指定端口，默認5672
-
cf.setPort(AMQP.PROTOCOL.PORT);
-
// 獲取一個新的連接
-
connection = cf.newConnection();
-
// 創建一個通道
-
channel = connection.createChannel();
-
-
//關閉管道和連接
-
channel.close();
-
connection.close();
2.2、聲明隊列
-
/**
-
* 申明一個隊列，如果這個隊列不存在，將會被創建
-
* @param queue 隊列名稱
-
* @param durable 持久性：true隊列會再重啟過后存在，但是其中的消息不會存在。
-
* @param exclusive 是否只能由創建者使用，其他連接不能使用。
-
* @param autoDelete 是否自動刪除（沒有連接自動刪除）
-
* @param arguments 隊列的其他屬性(構造參數)
-
* @return Queue.DeclareOk：宣告隊列的聲明確認方法已成功聲明。
-
* @throws java.io.IOException if an error is encountered
-
*/
-
channel.queueDeclare(
"testQueue",
true,
false,
false,
null);
此方法一般由Producer調用創建消息隊列。如果由Consumer創建隊列，有可能Producer發布消息的時候Queue還沒有被創建好，會造成消息丟失的情況。
2.3、聲明Exchange
-
/**
-
* 聲明一個 exchange.
-
* @param exchange 名稱
-
* @param type exchange type：direct、fanout、topic、headers
-
* @param durable 持久化
-
* @param autoDelete 是否自動刪除（沒有連接自動刪除）
-
* @param arguments 隊列的其他屬性(構造參數)
-
* @return 成功地聲明了一個聲明確認方法來指示交換。
-
* @throws java.io.IOException if an error is encountered
-
*/
-
channel.exchangeDeclare(
"leitao",
"topic",
true,
false,
null);
2.4、將queue和Exchange進行綁定（Binding）
-
/**
-
* 將隊列綁定到Exchange，不需要額外的參數。
-
* @param queue 隊列名稱
-
* @param exchange 交換機名稱
-
* @param routingKey 路由關鍵字
-
* @return Queue.BindOk：如果成功創建綁定，則返回綁定確認方法。
-
* @throws java.io.IOException if an error is encountered
-
*/
-
channel.queueBind(
"testQueue",
"leitao",
"testRoutingKey");
2.5、發布消息
-
/**
-
* 發布一條不用持久化的消息，且設置兩個監聽。
-
* @param exchange 消息交換機名稱,空字符串將使用直接交換器模式，發送到默認的Exchange=amq.direct。此狀態下，RoutingKey默認和Queue名稱相同
-
* @param routingKey 路由關鍵字
-
* @param mandatory 監聽是否有符合的隊列
-
* @param immediate 監聽符合的隊列上是有至少一個Consumer
-
* @param BasicProperties 設置消息持久化：MessageProperties.PERSISTENT_TEXT_PLAIN是持久化；MessageProperties.TEXT_PLAIN是非持久化。
-
* @param body 消息對象轉換的byte[]
-
* @throws java.io.IOException if an error is encountered
-
*/
-
channel.basicPublish(
"",queueName,
true,
false,MessageProperties.TEXT_PLAIN,SerializationUtils.serialize(object));
當exchange的值為空字符串或者是amq.direct時，此時的交換器類型默認是direct類型，可以不用單獨聲明Exchange，也不用單獨進行Binding，系統默認將queue名稱作為RoutingKey進行了綁定。
兩個傳入參數的含義
mandatory
當mandatory標志位設置為true時，如果exchange根據自身類型和消息routeKey無法找到一個符合條件的queue，那么會調用basic.return方法將消息返回給生產者（Basic.Return + Content-Header + Content-Body）；當mandatory設置為false時，出現上述情形broker會直接將消息扔掉。
immediate
當immediate標志位設置為true時，如果exchange在將消息路由到queue(s)時發現對於的queue上沒有消費者，那么這條消息不會放入隊列中。當與消息routeKey關聯的所有queue（一個或者多個）都沒有消費者時，該消息會通過basic.return方法返還給生產者。
概括來說，mandatory標志告訴服務器至少將該消息route到一個隊列中，否則將消息返還給生產者；immediate標志告訴服務器如果該消息關聯的queue上有消費者，則馬上將消息投遞給它，如果所有queue都沒有消費者，直接把消息返還給生產者，不用將消息入隊列等待消費者了。
注意：在RabbitMQ3.0以后的版本里，去掉了immediate參數的支持，發送帶immediate=true標記的publish會返回如下錯誤：
com.rabbitmq.client.AlreadyClosedException: connection is already closed due to connection error;protocol method: #method<connection.close>(reply-code=540, reply-text=NOT_IMPLEMENTED - immediate=true, class-id=60, method-id=40)。
為什么取消支持：immediate標記會影響鏡像隊列性能，增加代碼復雜性，並建議采用“TTL”和“DLX”等方式替代。
2.6、接收消息
-
/**
-
* 設置消費批量投遞數目，一次性投遞10條消息。當消費者未確認消息累計達到10條時，rabbitMQ將不會向此Channel上的消費者投遞消息，直到未確認數小於10條再投遞
-
* @param prefetchCount 投遞數目
-
* @param global 是否針對整個Channel。true表示此投遞數是給Channel設置的，false是給Channel上的Consumer設置的。
-
* @throws java.io.IOException if an error is encountered
-
*/
-
channel.basicQos(
10,
false);
-
//整個傳輸管道最多15條，具體分到每個消費者身上又不能大於10條
-
channel.basicQos(
15,
true);
-
-
/**
-
* 開始一個非局部、非排他性消費， with a server-generated consumerTag.
-
* 執行這個方法會回調handleConsumeOk方法
-
* @param queue 隊列名稱
-
* @param autoAck 是否自動應答。false表示consumer在成功消費過后必須要手動回復一下服務器，如果不回復，服務器就將認為此條消息消費失敗，繼續分發給其他consumer。
-
* @param callback 回調方法類，一般為自己的Consumer類
-
* @return 由服務器生成的consumertag
-
* @throws java.io.IOException if an error is encountered
-
*/
-
channel.basicConsume(queueName,
false, Consumer);
2.7、Consumer處理消息
-
/**
-
* 消費者收到消息的回調函數
-
* @param consumerTag 消費者標簽
-
* @param envelope 消息的包裝數據
-
* @param properties 消息的內容頭數據
-
* @param body 消息對象的byte[]
-
* @throws IOException
-
*/
-
void handleDelivery(String consumerTag,
-
Envelope envelope,
-
AMQP.BasicProperties properties,
-
byte[] body)
-
throws IOException;
3、Producer消息確認機制
3.1、什么是生產者消息確認機制？
沒有消息確認模式時，生產者不知道消息是不是已經到達了Broker服務器，這對於一些業務嚴謹的系統來說將是災難性的。消息確認模式可以采用AMQP協議層面提供的事務機制實現（此文沒有這種實現方式），但是會降低RabbitMQ的吞吐量。RabbitMQ自身提供了一種更加高效的實現方式：confirm模式。
消息生產者通過調用Channel.confirmSelect()方法將Channel信道設置成confirm模式。一旦信道被設置成confirm模式，該信道上的所有消息都會被指派一個唯一的ID（從1開始），一旦消息被對應的Exchange接收，Broker就會發送一個確認給生產者（其中deliveryTag就是此唯一的ID），這樣消息生產者就知道消息已經成功到達Broker。
confirm模式最大的好處在於他是異步的，一旦發布一條消息，生產者應用程序就可以在等信道返回確認的同時繼續發送下一條消息，當消息最終得到確認之后，生產者應用便可以通過回調方法來處理該確認消息，如果RabbitMQ因為自身內部錯誤導致消息丟失，就會發送一條nack消息，生產者應用程序同樣可以在回調方法中處理該nack消息。
在channel 被設置成 confirm 模式之后，所有被 publish 的后續消息都將被 confirm（即 ack） 或者被nack一次。但是沒有對消息被 confirm 的快慢做任何保證，並且同一條消息不會既被 confirm又被nack 。
3.2、開啟confirm模式
如上所說生產者通過調用Channel.confirmSelect()方法將Channel信道設置成confirm模式。
注意：已經在transaction事務模式的channel是不能再設置成confirm模式的，即這兩種模式是不能共存的。
3.3、普通confirm模式
普通confirm模式是串行的，即每次發送了一次消息，生產者都要等待Broker的確認消息，然后根據確認標記權衡消息重發還是繼續發下一條。由於是串行的，在效率上是比較低下的。
（1）重點方法
-
/**
-
* 等待Broker返回消息確認標記
-
* 注意，在非確定的通道，waitforconfirms拋出IllegalStateException。
-
* @return 是否發送成功
-
* @throws java.lang.IllegalStateException
-
*/
-
boolean waitForConfirms() throws InterruptedException;
（2 ） 部分使用代碼如下：
-
//注意：返回的時候Return在前，Confirm在后
-
channel.confirmSelect();
-
int i=
1;
-
while (i<=
50) {
-
//發布消息
-
channel.basicPublish(
"",queueName,
true,MessageProperties.TEXT_PLAIN,SerializationUtils.serialize(object));
-
//等待Broker的確認回調
-
if(channel.waitForConfirms())
-
System.out.println(
"send success!");
-
else
-
System.out.println(
"send error!");
-
i++;
-
}
3.4、批量confirm模式
批量confirm模式是異步的方式，效率要比普通confirm模式高許多，但是此種方式也會造成線程阻塞，想要進行失敗重發就必須要捕獲異常。網絡上還有采用waitForConfirms()實現批量confirm模式的，但是只要一條失敗了，就必須把這批次的消息統統再重發一次，非常的消耗性能，因此此文不予考慮。
（1）重點代碼
-
/**
-
* 等待直到所有消息被確認或者某個消息發送失敗。如果消息發送確認失敗了，
-
* waitForConfirmsOrDie 會拋出IOException異常。當在非確認通道上調用時
-
* ，會拋出IllegalStateException異常。
-
* @throws java.lang.IllegalStateException
-
*/
-
void waitForConfirmsOrDie() throws IOException, InterruptedException;
（2 ）部分代碼如下：
-
//注意：返回的時候Return在前，Confirm在后
-
channel.confirmSelect();
-
int i=
1;
-
while (i<=
50) {
-
//發布消息
-
channel.basicPublish(
"",queueName,
true,MessageProperties.TEXT_PLAIN,SerializationUtils.serialize(object));
-
i++;
-
}
-
channel.waitForConfirmsOrDie();
3.5、ConfirmListener監聽器模式
RabbitMQ提供了一個ConfirmListener接口專門用來進行確認監聽，我們可以實現ConfirmListener接口來創建自己的消息確認監聽。ConfirmListener接口中包含兩個回調方法：
-
/**
-
* 生產者發送消息到exchange成功的回調方法
-
*/
-
void handleAck(long deliveryTag, boolean multiple) throws IOException;
-
/**
-
* 生產者發送消息到服務器broker失敗的回調方法，服務器丟失了此消息。
-
* 注意，丟失的消息仍然可以傳遞給消費者，但broker不能保證這一點。
-
*/
-
void handleNack(long deliveryTag, boolean multiple) throws IOException;
其中deliveryTag是Broker給每條消息指定的唯一ID（從1開始）；multiple表示是否接收所有的應答消息，比如multiple=true時，發送100條消息成功過后，我們並不會收到100次handleAck方法調用。
（1）重要方法
-
//注冊消息確認監聽器
-
channel.addConfirmListener(
new MyConfirmListener());
（2）部分使用代碼如下：
-
//注意：返回的時候Return在前，Confirm在后
-
channel.confirmSelect();
-
//注冊消息確認監聽器
-
channel.addConfirmListener(
new MyConfirmListener());
-
//注冊消息結果返回監聽器
-
channel.addReturnListener(
new MyReturnListener());
-
int i=
1;
-
while (i<=
50) {
-
//發布消息
-
channel.basicPublish(
"",queueName,
true,MessageProperties.TEXT_PLAIN,SerializationUtils.
-
serialize(object));
-
i++;
-
}
-
//自定義的消息確認監聽器
-
public
class MyConfirmListener implements ConfirmListener{
-
/**
-
* 生產者發送消息到exchange成功的回調方法
-
* 消息被Exchange接受以后，如果沒有匹配的Queue，則會被丟棄。但是可以設置ReturnListener監聽來監聽有沒有匹配的隊列。
-
* 因此handleAck執行了，並不能完全表示消息已經進入了對應的隊列，只能表示對應的exchange成功的接收了消息。
-
* 消息被exchange接收過后，還需要通過一定的匹配規則分發到對應的隊列queue中。
-
*/
-
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
-
//注意：deliveryTag是broker給消息指定的唯一id（從1開始）
-
System.out.println(
"Exchange接收消息："+deliveryTag+
"（deliveryTag）成功！multiple="+multiple);
-
}
-
/**
-
* 生產者發送消息到服務器broker失敗的回調方法，服務器丟失了此消息。
-
* 注意，丟失的消息仍然可以傳遞給消費者，但broker不能保證這一點。（不明白，既然丟失了，為啥還能發送）
-
*/
-
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
-
System.out.println(
"Exchange接收消息："+deliveryTag+
"（deliveryTag）失敗！服務器broker丟失了消息");
-
}
-
}
-
//自定義的結果返回監聽器
-
/**
-
* 實現此接口以通知交付basicpublish失敗時，“mandatory”或“immediate”的標志監聽（源代碼注釋翻譯）。
-
* 在發布消息時設置mandatory等於true，監聽消息是否有相匹配的隊列，
-
* 沒有時ReturnListener將執行handleReturn方法，消息將返給發送者
-
*/
-
public
class MyReturnListener implements ReturnListener {
-
public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,
-
BasicProperties properties, byte[] body)
throws IOException {
-
System.out.println(
"消息發送到隊列失敗：回復失敗編碼："+replyCode+
";回復失敗文本："+replyText+
";失敗消息對象："+SerializationUtils.deserialize(body));
-
}
-
}
4、Consumer消息確認機制
為了保證消息從隊列可靠地到達消費者，RabbitMQ提供消息確認機制(message acknowledgment)。消費者在注冊消費者時，可以指定noAck參數，當noAck=false時，RabbitMQ會等待消費者顯式發回ack信號后才從內存（或磁盤，如果是持久化消息的話）中移去消息。否則，RabbitMQ會在隊列中消息被消費后立即刪除它。
當noAck=false時，對於RabbitMQ服務器端而言，隊列中的消息分成了兩部分：一部分是等待投遞給消費者的消息（web管理界面上的Ready狀態）；一部分是已經投遞給消費者，但是還沒有收到消費者ack信號的消息（web管理界面上的Unacked狀態）。如果服務器端一直沒有收到消費者的ack信號，並且消費此消息的消費者已經斷開連接，則服務器端會安排該消息重新進入隊列，等待投遞給下一個消費者（也可能還是原來的那個消費者）。
（1）重要方法
-
/**
-
*1. 開始一個非局部、非排他性消費， with a server-generated consumerTag.
-
* 注意：執行這個方法會回調handleConsumeOk方法,在此方法中處理消息。
-
* @param queue 隊列名稱
-
* @param autoAck 是否自動應答。false表示consumer在成功消費過后必須要手動回復一下服務器，如果不回復，服務器就將認為此條消息消費失敗，繼續分發給其他consumer。
-
* @param callback 回調方法類
-
* @return 由服務器生成的consumertag
-
* @throws java.io.IOException if an error is encountered
-
*/
-
String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;
-
-
-
/**
-
*2
-
consumer處理成功后，通知broker刪除隊列中的消息，如果設置multiple=true，表示支持批量確認機制以減少網絡流量。
-
例如：有值為5,6,7,8 deliveryTag的投遞
-
如果此時channel.basicAck(8, true);則表示前面未確認的5,6,7投遞也一起確認處理完畢。
-
如果此時channel.basicAck(8, false);則僅表示deliveryTag=8的消息已經成功處理。
-
*/
-
void basicAck(long deliveryTag, boolean multiple) throws IOException;
-
-
/**3
-
consumer處理失敗后，例如：有值為5,6,7,8 deliveryTag的投遞。
-
如果channel.basicNack(8, true, true);表示deliveryTag=8之前未確認的消息都處理失敗且將這些消息重新放回隊列中。
-
如果channel.basicNack(8, true, false);表示deliveryTag=8之前未確認的消息都處理失敗且將這些消息直接丟棄。
-
如果channel.basicNack(8, false, true);表示deliveryTag=8的消息處理失敗且將該消息重新放回隊列。
-
如果channel.basicNack(8, false, false);表示deliveryTag=8的消息處理失敗且將該消息直接丟棄。
-
*/
-
void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;
-
-
/**4
-
相比channel.basicNack，除了沒有multiple批量確認機制之外，其他語義完全一樣。
-
如果channel.basicReject(8, true);表示deliveryTag=8的消息處理失敗且將該消息重新放回隊列。
-
如果channel.basicReject(8, false);表示deliveryTag=8的消息處理失敗且將該消息直接丟棄。
-
*/
-
void basicReject(long deliveryTag, boolean requeue) throws IOException;
（2）部分使用代碼如下：
-
//this表示自己的Consumer
-
channel.basicConsume(queueName,
false,
this);
-
...
-
@Override
-
public void handleDelivery(String arg0, Envelope envelope, BasicProperties arg2, byte[] body) throws IOException {
-
if (body ==
null)
-
return;
-
Map<String, Object> map = (Map<String, Object>) SerializationUtils.deserialize(body);
-
/**
-
* 專門處理奇數消息的消費者
-
*/
-
int tagId = (Integer) map.get(
"tagId");
-
if (tagId %
2 !=
0) {
-
//處理消息
-
System.out.println(
"接收並處理消息："+tagId);
-
//通知服務器此消息已經被處理了
-
channel.basicAck(envelope.getDeliveryTag(),
false);
-
}
else{
-
//通知服務器消息處理失敗，重新放回隊列。false表示處理失敗消息不放會隊列，直接刪除
-
channel.basicReject(envelope.getDeliveryTag(),
true);
-
}
-
}
5、Demo項目整體代碼
此demo就是向RabbitMQ服務器上面發送20個消息，消息體是map，里面裝的是tagId=數字。然后注冊了兩個消費者，分別處理奇數和偶數。
5.1、連接工具類
-
/**
-
* 連接工具類
-
*/
-
public
class ConnectionUtil {
-
-
Channel channel;
-
Connection connection;
-
String queueName;
-
-
public ConnectionUtil(String queueName) throws IOException {
-
this.queueName = queueName;
-
// 創建連接工廠
-
ConnectionFactory cf =
new ConnectionFactory();
-
// 設置rabbitmq服務器IP地址
-
cf.setHost(
"*.16.0.*");
-
// 設置rabbitmq服務器用戶名
-
cf.setUsername(
"*");
-
// 設置rabbitmq服務器密碼
-
cf.setPassword(
"*");
-
cf.setPort(AMQP.PROTOCOL.PORT);
-
// 獲取一個新的連接
-
connection = cf.newConnection();
-
// 創建一個通道
-
channel = connection.createChannel();
-
/**
-
*申明一個隊列，如果這個隊列不存在，將會被創建
-
* @param queue 隊列名稱
-
* @param durable 持久性：true隊列會再重啟過后存在，但是其中的消息不會存在。
-
* @param exclusive 是否只能由創建者使用
-
* @param autoDelete 是否自動刪除（沒有連接自動刪除）
-
* @param arguments 隊列的其他屬性(構造參數)
-
* @return 宣告隊列的聲明確認方法已成功聲明。
-
* @throws java.io.IOException if an error is encountered
-
*/
-
channel.queueDeclare(queueName,
true,
false,
false,
null);
-
}
-
-
public void close() throws IOException{
-
channel.close();
-
connection.close();
-
}
-
}
5.2、具體生產者
-
/**
-
* 消息生產者
-
*/
-
public
class MessageProducer {
-
-
private ConnectionUtil connectionUtil;
-
-
public MessageProducer(ConnectionUtil connectionUtil){
-
this.connectionUtil=connectionUtil;
-
}
-
/**
-
* 發送消息到隊列中
-
*/
-
public void sendMessage(Serializable object) throws IOException{
-
/**
-
* Publish a message
-
* @param exchange 消息交換機名稱,空字符串將使用直接交換器模式，發送到默認的Exchange=amq.direct
-
* @param routingKey 路由關鍵字
-
* @param mandatory 監聽是否有符合的隊列
-
* @param BasicProperties 設置消息持久化：MessageProperties.PERSISTENT_TEXT_PLAIN是持久化；MessageProperties.TEXT_PLAIN是非持久化
-
* @param body 消息對象
-
* @throws java.io.IOException if an error is encountered
-
*/
-
connectionUtil.channel.basicPublish(
"", connectionUtil.queueName,
true, MessageProperties.TEXT_PLAIN, SerializationUtils.serialize(object));
-
System.out.println(
"MessageProducer發送了一條消息："+object);
-
}
-
}
5.3、公共消費者父類
-
/**
-
* 消息消費者基礎類
-
*/
-
public
class MessageConsumer implements Consumer {
-
//消費者標簽，注冊成功時由rabbitmq服務器自動生成
-
protected String consumerTag;
-
-
protected ConnectionUtil connectionUtil;
-
-
public MessageConsumer(ConnectionUtil connectionUtil){
-
this.connectionUtil=connectionUtil;
-
}
-
-
public void basicConsume(){
-
try {
-
/**
-
* 設置消費投遞數目，一次性投遞10條消息。當消費者未確認消息達到10條時，rabbitMQ將不會向此消費者投遞消息，直到未確認數小於10條再投遞
-
* @param prefetchCount 投遞數目
-
* @param global 是否針對整個Channel。true表示此投遞數是給Channel設置的，false是給Channel上的Consumer設置的。
-
* @throws java.io.IOException if an error is encountered
-
*/
-
connectionUtil.channel.basicQos(
10,
false);
//表示每個消費者最多10條
-
connectionUtil.channel.basicQos(
15,
true);
//整個傳輸管道最多15條，具體分到每個消費者身上又不能大於10條
-
/**
-
* 開始一個非局部、非排他性消費， with a server-generated consumerTag.
-
* 執行這個方法會回調handleConsumeOk方法
-
* @param queue 隊列名稱
-
* @param autoAck 是否自動應答。false表示consumer在成功消費過后必須要手動回復一下服務器，如果不回復，服務器就將認為此條消息消費失敗，繼續分發給其他consumer。
-
* @param callback 回調方法類
-
* @return 由服務器生成的consumertag
-
* @throws java.io.IOException if an error is encountered
-
*/
-
connectionUtil.channel.basicConsume(connectionUtil.queueName,
false,
this);
-
}
catch (IOException e) {
-
e.printStackTrace();
-
}
-
}
-
-
/**
-
* 收到消息時的回調函數
-
*/
-
public void handleDelivery(String arg0, Envelope arg1, BasicProperties arg2, byte[] arg3) throws IOException {
-
//子類重寫覆蓋具體操作
-
}
-
-
/**
-
* 消費者注冊成功回調函數
-
*/
-
public void handleConsumeOk(String consumerTag) {
-
this.consumerTag=consumerTag;
-
System.out.println(
"消費者："+consumerTag+
",注冊成功！");
-
}
-
-
/**
-
* 手動取消消費者注冊成功回調函數
-
* 當調用Channel類的void basicCancel(String consumerTag) throws IOException;方法觸發此回調函數
-
*/
-
public void handleCancelOk(String consumerTag) {
-
System.out.println(consumerTag+
" 手動取消消費者注冊成功!");
-
}
-
-
/**
-
* 當消費者因為其他原因被動取消注冊時調用，比如queue被刪除了。
-
*/
-
public void handleCancel(String consumerTag) throws IOException {
-
System.out.println(
"因為外部原因消費者："+consumerTag+
" 取消注冊！");
-
}
-
-
/**
-
* 當通道或基礎連接被關閉時調用
-
*/
-
public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
-
System.out.println(
"通道或基礎連接被關閉");
-
}
-
-
/**
-
* Called when a <code><b>basic.recover-ok</b></code> is received
-
* in reply to a <code><b>basic.recover</b></code>. All messages
-
* received before this is invoked that haven't been <i>ack</i>'ed will be
-
* re-delivered. All messages received afterwards won't be.
-
* @param consumerTag the <i>consumer tag</i> associated with the consumer
-
*/
-
public void handleRecoverOk(String consumerTag) {
-
-
}
-
}
5.4、具體的消費者
-
/**
-
* 專門處理偶數消息的消費者
-
*/
-
public
class EvenConsumer extends MessageConsumer {
-
-
public EvenConsumer(ConnectionUtil connectionUtil) {
-
super(connectionUtil);
-
}
-
-
@Override
-
public void handleConsumeOk(String consumerTag) {
-
this.consumerTag=consumerTag;
-
System.out.println(
"EvenConsumer消費者："+consumerTag+
",注冊成功！");
-
}
-
-
@Override
-
public void handleDelivery(String arg0, Envelope envelope, BasicProperties arg2, byte[] body) throws IOException {
-
if (body ==
null)
-
return;
-
Map<String, Object> map = (Map<String, Object>) SerializationUtils.deserialize(body);
-
int tagId = (Integer) map.get(
"tagId");
-
if (tagId %
2 ==
0) {
-
//處理消息
-
System.out.println(
"EvenConsumer接收並處理消息："+tagId);
-
//通知服務器此消息已經被處理了
-
connectionUtil.channel.basicAck(envelope.getDeliveryTag(),
false);
-
}
else{
-
//通知服務器消息處理失敗，重新放回隊列。false表示處理失敗消息不放會隊列，直接刪除
-
connectionUtil.channel.basicReject(envelope.getDeliveryTag(),
true);
-
}
-
}
-
}
-
/**
-
* 專門處理奇數消息的消費者
-
*/
-
public
class OddConsumer extends MessageConsumer {
-
-
public OddConsumer(ConnectionUtil connectionUtil) {
-
super(connectionUtil);
-
}
-
-
@Override
-
public void handleConsumeOk(String consumerTag) {
-
this.consumerTag=consumerTag;
-
System.out.println(
"OddConsumer消費者："+consumerTag+
",注冊成功！");
-
}
-
-
@Override
-
public void handleDelivery(String arg0, Envelope envelope, BasicProperties arg2, byte[] body) throws IOException {
-
if (body ==
null)
-
return;
-
Map<String, Object> map = (Map<String, Object>) SerializationUtils.deserialize(body);
-
int tagId = (Integer) map.get(
"tagId");
-
if (tagId %
2 !=
0) {
-
//處理消息
-
System.out.println(
"OddConsumer接收並處理消息："+tagId);
-
//通知服務器此消息已經被處理了
-
connectionUtil.channel.basicAck(envelope.getDeliveryTag(),
false);
-
}
else{
-
//通知服務器消息處理失敗，重新放回隊列。false表示處理失敗消息不放會隊列，直接刪除
-
connectionUtil.channel.basicReject(envelope.getDeliveryTag(),
true);
-
}
-
}
-
}
5.5、監聽器
-
/**
-
*producer發送確認事件。
-
*/
-
public
class MyConfirmListener implements ConfirmListener{
-
/**
-
* 生產者發送消息到exchange成功的回調方法
-
* 消息被Exchange接受以后，如果沒有匹配的Queue，則會被丟棄。但是可以設置ReturnListener監聽來監聽有沒有匹配的隊列。
-
* 因此handleAck執行了，並不能完全表示消息已經進入了對應的隊列，只能表示對應的exchange成功的接收了消息。
-
* 消息被exchange接收過后，還需要通過一定的匹配規則分發到對應的隊列queue中。
-
*/
-
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
-
//注意：deliveryTag是broker給消息指定的唯一id（從1開始）
-
System.out.println(
"Exchange接收消息："+deliveryTag+
"（deliveryTag）成功！multiple="+multiple);
-
}
-
/**
-
* 生產者發送消息到服務器broker失敗的回調方法，服務器丟失了此消息。
-
* 注意，丟失的消息仍然可以傳遞給消費者，但broker不能保證這一點。
-
*/
-
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
-
System.out.println(
"Exchange接收消息："+deliveryTag+
"（deliveryTag）失敗！服務器broker丟失了消息");
-
}
-
}
-
/**
-
* 實現此接口以通知交付basicpublish失敗時，“mandatory”或“immediate”的標志監聽（源代碼注釋翻譯）。
-
* 在發布消息時設置mandatory等於true，監聽消息是否有相匹配的隊列，
-
* 沒有時ReturnListener將執行handleReturn方法，消息將返給發送者 。
-
* 由於3.0版本過后取消了支持immediate，此處不做過多的解釋。
-
*/
-
public
class MyReturnListener implements ReturnListener {
-
-
public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,
-
BasicProperties properties, byte[] body)
throws IOException {
-
System.out.println(
"消息發送到隊列失敗：回復失敗編碼："+replyCode+
";回復失敗文本："+replyText+
";失敗消息對象："+SerializationUtils.deserialize(body));
-
}
-
}
5.6、客戶端
-
public
class Client {
-
-
public static void main(String[] args) {
-
new Client();
-
}
-
-
public Client(){
-
try {
-
//發消息
-
publishMessage();
-
//注冊消費者
-
addConsumer();
-
}
catch (IOException e) {
-
e.printStackTrace();
-
}
catch (InterruptedException e) {
-
e.printStackTrace();
-
}
-
}
-
-
public void publishMessage() throws IOException, InterruptedException{
-
ConnectionUtil connectionUtil=
new ConnectionUtil(
"testqueue");
-
MessageProducer producer=
new MessageProducer(connectionUtil);
-
connectionUtil.channel.confirmSelect();
-
//注意：返回的時候Return在前，Confirm在后
-
connectionUtil.channel.addConfirmListener(
new MyConfirmListener());
-
connectionUtil.channel.addReturnListener(
new MyReturnListener());
-
int i=
1;
-
while (i<=
10) {
-
HashMap<String, Object> map=
new HashMap<String, Object>();
-
map.put(
"tagId", i);
-
producer.sendMessage(map);
-
i++;
-
}
-
}
-
-
public void addConsumer() throws IOException{
-
ConnectionUtil connectionUtil=
new ConnectionUtil(
"testqueue");
-
OddConsumer odd=
new OddConsumer(connectionUtil);
-
odd.basicConsume();
-
EvenConsumer even=
new EvenConsumer(connectionUtil);
-
even.basicConsume();
-
}
-
-
}
5.7、測試結果
-
MessageProducer發送了一條消息：{tagId=
1}
-
MessageProducer發送了一條消息：{tagId=
2}
-
MessageProducer發送了一條消息：{tagId=
3}
-
Exchange接收消息：
1（deliveryTag）成功！multiple=
false
-
Exchange接收消息：
2（deliveryTag）成功！multiple=
false
-
MessageProducer發送了一條消息：{tagId=
4}
-
Exchange接收消息：
3（deliveryTag）成功！multiple=
false
-
MessageProducer發送了一條消息：{tagId=
5}
-
Exchange接收消息：
4（deliveryTag）成功！multiple=
false
-
MessageProducer發送了一條消息：{tagId=
6}
-
Exchange接收消息：
5（deliveryTag）成功！multiple=
false
-
MessageProducer發送了一條消息：{tagId=
7}
-
Exchange接收消息：
6（deliveryTag）成功！multiple=
false
-
MessageProducer發送了一條消息：{tagId=
8}
-
Exchange接收消息：
7（deliveryTag）成功！multiple=
false
-
Exchange接收消息：
8（deliveryTag）成功！multiple=
false
-
MessageProducer發送了一條消息：{tagId=
9}
-
Exchange接收消息：
9（deliveryTag）成功！multiple=
false
-
MessageProducer發送了一條消息：{tagId=
10}
-
Exchange接收消息：
10（deliveryTag）成功！multiple=
false
-
OddConsumer消費者：amq.ctag-z8s8LaSgYvo02jktCZrCYA,注冊成功！
-
OddConsumer接收並處理消息：
1
-
OddConsumer接收並處理消息：
3
-
OddConsumer接收並處理消息：
5
-
OddConsumer接收並處理消息：
7
-
OddConsumer接收並處理消息：
9
-
EvenConsumer消費者：amq.ctag-LpN6Q5VvNY3wCof2lXqS4A,注冊成功！
-
EvenConsumer接收並處理消息：
4
-
EvenConsumer接收並處理消息：
8
-
EvenConsumer接收並處理消息：
2
-
EvenConsumer接收並處理消息：
10
-
EvenConsumer接收並處理消息：
6
-