RabbitMQ有以下幾種工作模式 ：
1、Work queues
2、Publish/Subscribe
3、Routing
4、Topics
5、Header
6、RPC

一.Work queues

work queues與入門程序HelloWord相比，多了一個消費端，兩個消費端共同消費同一個隊列中的消息。

應用場景：對于任務過重或任務較多情況使用工作隊列可以提高任務處理的速度。

測試：
1、使用入門程序HelloWord，啟動多個消費者。
2、生產者發送多個消息。

結果：
1、一條消息只會被一個消費者接收；
2、rabbit采用輪詢的方式將消息是平均發送給消費者的；
3、消費者在處理完某條消息后，才會收到下一條消息。

二.Publish/subscribe

1.工作模式

發布訂閱模式：
1、每個消費者監聽自己的隊列。
2、生產者將消息發給broker，由交換機將消息轉發到綁定此交換機的每個隊列，每個綁定交換機的隊列都將接收
到消息

2.代碼

案例：
當用戶充值成功或轉賬完成系統通知用戶，通知方式有短信、郵件多種方法 。

1）生產者

聲明Exchange_fanout_inform交換機。
聲明兩個隊列并且綁定到此交換機，綁定時不需要指定routingkey
發送消息時不需要指定routingkey

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer02_publish {//隊列名稱private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String QUEUE_INFORM_SMS = "queue_inform_sms";private static final String EXCHANGE_FANOUT_INFORM="exchange_fanout_inform";public static void main(String[] args) {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新連接connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成channel = connection.createChannel();//聲明隊列，如果隊列在mq 中沒有則要創建//參數：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//聲明一個交換機//參數：String exchange, String type/*** 參數明細：* 1、交換機的名稱* 2、交換機的類型* fanout：對應的rabbitmq的工作模式是 publish/subscribe* direct：對應的Routing	工作模式* topic：對應的Topics工作模式* headers： 對應的headers工作模式*/channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);//進行交換機和隊列綁定//參數：String queue, String exchange, String routingKey/*** 參數明細：* 1、queue 隊列名稱* 2、exchange 交換機名稱* 3、routingKey 路由key，作用是交換機根據路由key的值將消息轉發到指定的隊列中，在發布訂閱模式中調協為空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");//發送消息//參數：String exchange, String routingKey, BasicProperties props, byte[] body/*** 參數明細：* 1、exchange，交換機，如果不指定將使用mq的默認交換機（設置為""）* 2、routingKey，路由key，交換機根據路由key來將消息轉發到指定的隊列，如果使用默認交換機，routingKey設置為隊列的名稱* 3、props，消息的屬性* 4、body，消息內容*/for(int i=0;i<5;i++){//消息內容String message = "send inform message to user";channel.basicPublish(EXCHANGE_FANOUT_INFORM,"",null,message.getBytes());System.out.println("send to mq "+message);}} catch (Exception e) {e.printStackTrace();} finally {//關閉連接//先關閉通道try {channel.close();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}try {connection.close();} catch (IOException e) {e.printStackTrace();}}}
}

2）消費者

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer02_subscribe_email {//隊列名稱private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String EXCHANGE_FANOUT_INFORM="exchange_fanout_inform";public static void main(String[] args) throws IOException, TimeoutException {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");//建立新連接Connection connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成Channel channel = connection.createChannel();/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);//聲明一個交換機//參數：String exchange, String type/*** 參數明細：* 1、交換機的名稱* 2、交換機的類型* fanout：對應的rabbitmq的工作模式是 publish/subscribe* direct：對應的Routing	工作模式* topic：對應的Topics工作模式* headers： 對應的headers工作模式*/channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);//進行交換機和隊列綁定//參數：String queue, String exchange, String routingKey/*** 參數明細：* 1、queue 隊列名稱* 2、exchange 交換機名稱* 3、routingKey 路由key，作用是交換機根據路由key的值將消息轉發到指定的隊列中，在發布訂閱模式中調協為空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");//實現消費方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 當接收到消息后此方法將被調用* @param consumerTag  消費者標簽，用來標識消費者的，在監聽隊列時設置channel.basicConsume* @param envelope 信封，通過envelope* @param properties 消息屬性* @param body 消息內容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交換機String exchange = envelope.getExchange();//消息id，mq在channel中用來標識消息的id，可用于確認消息已接收long deliveryTag = envelope.getDeliveryTag();//消息內容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//監聽隊列//參數：String queue, boolean autoAck, Consumer callback/*** 參數明細：* 1、queue 隊列名稱* 2、autoAck 自動回復，當消費者接收到消息后要告訴mq消息已接收，如果將此參數設置為tru表示會自動回復mq，如果設置為false要通過編程實現回復* 3、callback，消費方法，當消費者接收到消息要執行的方法*/channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);}
}

3.測試

打開RabbitMQ的管理界面，觀察交換機綁定情況：
使用生產者發送若干條消息，每條消息都轉發到各個隊列，并且每個消費者都接收到了消息。

4.思考

1、publish/subscribe與work queues有什么異同。
區別：
1）work queues不用定義交換機，而publish/subscribe需要定義交換機。
2）publish/subscribe的生產方是面向交換機發送消息，work queues的生產方是面向隊列發送消息(底層使用默認交換機)。
3）publish/subscribe需要設置隊列和交換機的綁定，work queues不需要設置，實質上work queues會將隊列綁定到默認的交換機 。
相同點：
所以兩者實現的發布/訂閱的效果是一樣的，多個消費端監聽同一個隊列不會重復消費消息

2、實質工作用什么 publish/subscribe還是work queues。
建議使用 publish/subscribe，發布訂閱模式比工作隊列模式更強大，并且發布訂閱模式可以指定自己專用的交換
機

三.Routing

1.工作模式

路由模式：
1、每個消費者監聽自己的隊列，并且設置routingkey。
2、生產者將消息發給交換機，由交換機根據routingkey來轉發消息到指定的隊列。

2.代碼

聲明exchange_routing_inform交換機。
聲明兩個隊列并且綁定到此交換機，綁定時需要指定routingkey
發送消息時需要指定routingkey

1）生產者

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer03_routing {//隊列名稱private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String QUEUE_INFORM_SMS = "queue_inform_sms";private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";private static final String ROUTINGKEY_EMAIL="inform_email";private static final String ROUTINGKEY_SMS="inform_sms";public static void main(String[] args) {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新連接connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成channel = connection.createChannel();//聲明隊列，如果隊列在mq 中沒有則要創建//參數：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//聲明一個交換機//參數：String exchange, String type/*** 參數明細：* 1、交換機的名稱* 2、交換機的類型* fanout：對應的rabbitmq的工作模式是 publish/subscribe* direct：對應的Routing	工作模式* topic：對應的Topics工作模式* headers： 對應的headers工作模式*/channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);//進行交換機和隊列綁定//參數：String queue, String exchange, String routingKey/*** 參數明細：* 1、queue 隊列名稱* 2、exchange 交換機名稱* 3、routingKey 路由key，作用是交換機根據路由key的值將消息轉發到指定的隊列中，在發布訂閱模式中調協為空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,"inform");channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS);channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM,"inform");//發送消息//參數：String exchange, String routingKey, BasicProperties props, byte[] body/*** 參數明細：* 1、exchange，交換機，如果不指定將使用mq的默認交換機（設置為""）* 2、routingKey，路由key，交換機根據路由key來將消息轉發到指定的隊列，如果使用默認交換機，routingKey設置為隊列的名稱* 3、props，消息的屬性* 4、body，消息內容*/for(int i=0;i<5;i++){//發送消息的時候指定routingKeyString message = "send email inform message to user";channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL,null,message.getBytes());System.out.println("send to mq "+message);}for(int i=0;i<5;i++){//發送消息的時候指定routingKeyString message = "send sms inform message to user";channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS,null,message.getBytes());System.out.println("send to mq "+message);}for(int i=0;i<5;i++){//發送消息的時候指定routingKeyString message = "send inform message to user";channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform",null,message.getBytes());System.out.println("send to mq "+message);}} catch (Exception e) {e.printStackTrace();} finally {//關閉連接//先關閉通道try {channel.close();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}try {connection.close();} catch (IOException e) {e.printStackTrace();}}}
}

2）消費者

消費者一：郵件

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer03_routing_email {//隊列名稱private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";private static final String ROUTINGKEY_EMAIL="inform_email";public static void main(String[] args) throws IOException, TimeoutException {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");//建立新連接Connection connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成Channel channel = connection.createChannel();/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);//聲明一個交換機//參數：String exchange, String type/*** 參數明細：* 1、交換機的名稱* 2、交換機的類型* fanout：對應的rabbitmq的工作模式是 publish/subscribe* direct：對應的Routing	工作模式* topic：對應的Topics工作模式* headers： 對應的headers工作模式*/channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);//進行交換機和隊列綁定//參數：String queue, String exchange, String routingKey/*** 參數明細：* 1、queue 隊列名稱* 2、exchange 交換機名稱* 3、routingKey 路由key，作用是交換機根據路由key的值將消息轉發到指定的隊列中，在發布訂閱模式中調協為空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);//實現消費方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 當接收到消息后此方法將被調用* @param consumerTag  消費者標簽，用來標識消費者的，在監聽隊列時設置channel.basicConsume* @param envelope 信封，通過envelope* @param properties 消息屬性* @param body 消息內容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交換機String exchange = envelope.getExchange();//消息id，mq在channel中用來標識消息的id，可用于確認消息已接收long deliveryTag = envelope.getDeliveryTag();//消息內容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//監聽隊列//參數：String queue, boolean autoAck, Consumer callback/*** 參數明細：* 1、queue 隊列名稱* 2、autoAck 自動回復，當消費者接收到消息后要告訴mq消息已接收，如果將此參數設置為tru表示會自動回復mq，如果設置為false要通過編程實現回復* 3、callback，消費方法，當消費者接收到消息要執行的方法*/channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);}
}

消費者二：短信

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer03_routing_sms {//隊列名稱private static final String QUEUE_INFORM_SMS = "queue_inform_sms";private static final String EXCHANGE_ROUTING_INFORM="exchange_routing_inform";private static final String ROUTINGKEY_SMS="inform_sms";public static void main(String[] args) throws IOException, TimeoutException {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");//建立新連接Connection connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成Channel channel = connection.createChannel();/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//聲明一個交換機//參數：String exchange, String type/*** 參數明細：* 1、交換機的名稱* 2、交換機的類型* fanout：對應的rabbitmq的工作模式是 publish/subscribe* direct：對應的Routing	工作模式* topic：對應的Topics工作模式* headers： 對應的headers工作模式*/channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);//進行交換機和隊列綁定//參數：String queue, String exchange, String routingKey/*** 參數明細：* 1、queue 隊列名稱* 2、exchange 交換機名稱* 3、routingKey 路由key，作用是交換機根據路由key的值將消息轉發到指定的隊列中，在發布訂閱模式中調協為空字符串*/channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS);//實現消費方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 當接收到消息后此方法將被調用* @param consumerTag  消費者標簽，用來標識消費者的，在監聽隊列時設置channel.basicConsume* @param envelope 信封，通過envelope* @param properties 消息屬性* @param body 消息內容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交換機String exchange = envelope.getExchange();//消息id，mq在channel中用來標識消息的id，可用于確認消息已接收long deliveryTag = envelope.getDeliveryTag();//消息內容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//監聽隊列//參數：String queue, boolean autoAck, Consumer callback/*** 參數明細：* 1、queue 隊列名稱* 2、autoAck 自動回復，當消費者接收到消息后要告訴mq消息已接收，如果將此參數設置為tru表示會自動回復mq，如果設置為false要通過編程實現回復* 3、callback，消費方法，當消費者接收到消息要執行的方法*/channel.basicConsume(QUEUE_INFORM_SMS,true,defaultConsumer);}
}

3.測試

使用生產者發送若干條消息，交換機根據routingkey轉發消息到指定的隊列

4.思考

1、Routing模式和Publish/subscibe有什么區別？
Routing模式要求隊列在綁定交換機時要指定routingkey，消息會轉發到符合routingkey的隊列。

四.Topics

1.工作模式

路由模式：
1、每個消費者監聽自己的隊列，并且設置帶統配符的routingkey。
2、生產者將消息發給broker，由交換機根據routingkey來轉發消息到指定的隊列。

2.代碼

案例：
根據用戶的通知設置去通知用戶，設置接收Email的用戶只接收Email，設置接收sms的用戶只接收sms，設置兩種
通知類型都接收的則兩種通知都有效。

1）生產者

聲明交換機，指定topic類型

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer04_topics {//隊列名稱private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String QUEUE_INFORM_SMS = "queue_inform_sms";private static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";private static final String ROUTINGKEY_EMAIL="inform.#.email.#";private static final String ROUTINGKEY_SMS="inform.#.sms.#";public static void main(String[] args) {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新連接connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成channel = connection.createChannel();//聲明隊列，如果隊列在mq 中沒有則要創建//參數：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//聲明一個交換機//參數：String exchange, String type/*** 參數明細：* 1、交換機的名稱* 2、交換機的類型* fanout：對應的rabbitmq的工作模式是 publish/subscribe* direct：對應的Routing	工作模式* topic：對應的Topics工作模式* headers： 對應的headers工作模式*/channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);//進行交換機和隊列綁定//參數：String queue, String exchange, String routingKey/*** 參數明細：* 1、queue 隊列名稱* 2、exchange 交換機名稱* 3、routingKey 路由key，作用是交換機根據路由key的值將消息轉發到指定的隊列中，在發布訂閱模式中調協為空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_EMAIL);channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_SMS);//發送消息//參數：String exchange, String routingKey, BasicProperties props, byte[] body/*** 參數明細：* 1、exchange，交換機，如果不指定將使用mq的默認交換機（設置為""）* 2、routingKey，路由key，交換機根據路由key來將消息轉發到指定的隊列，如果使用默認交換機，routingKey設置為隊列的名稱* 3、props，消息的屬性* 4、body，消息內容*/for(int i=0;i<5;i++){//發送消息的時候指定routingKeyString message = "send email inform message to user";channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.email",null,message.getBytes());System.out.println("send to mq "+message);}for(int i=0;i<5;i++){//發送消息的時候指定routingKeyString message = "send sms inform message to user";channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.sms",null,message.getBytes());System.out.println("send to mq "+message);}for(int i=0;i<5;i++){//發送消息的時候指定routingKeyString message = "send sms and email inform message to user";channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.sms.email",null,message.getBytes());System.out.println("send to mq "+message);}} catch (Exception e) {e.printStackTrace();} finally {//關閉連接//先關閉通道try {channel.close();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}try {connection.close();} catch (IOException e) {e.printStackTrace();}}}
}

2）消費者

消費者一：email

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer04_topics_email {//隊列名稱private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";private static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";private static final String ROUTINGKEY_EMAIL="inform.#.email.#";public static void main(String[] args) throws IOException, TimeoutException {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");//建立新連接Connection connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成Channel channel = connection.createChannel();/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);//聲明一個交換機//參數：String exchange, String type/*** 參數明細：* 1、交換機的名稱* 2、交換機的類型* fanout：對應的rabbitmq的工作模式是 publish/subscribe* direct：對應的Routing	工作模式* topic：對應的Topics工作模式* headers： 對應的headers工作模式*/channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);//進行交換機和隊列綁定//參數：String queue, String exchange, String routingKey/*** 參數明細：* 1、queue 隊列名稱* 2、exchange 交換機名稱* 3、routingKey 路由key，作用是交換機根據路由key的值將消息轉發到指定的隊列中，在發布訂閱模式中調協為空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPICS_INFORM,ROUTINGKEY_EMAIL);//實現消費方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 當接收到消息后此方法將被調用* @param consumerTag  消費者標簽，用來標識消費者的，在監聽隊列時設置channel.basicConsume* @param envelope 信封，通過envelope* @param properties 消息屬性* @param body 消息內容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交換機String exchange = envelope.getExchange();//消息id，mq在channel中用來標識消息的id，可用于確認消息已接收long deliveryTag = envelope.getDeliveryTag();//消息內容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//監聽隊列//參數：String queue, boolean autoAck, Consumer callback/*** 參數明細：* 1、queue 隊列名稱* 2、autoAck 自動回復，當消費者接收到消息后要告訴mq消息已接收，如果將此參數設置為tru表示會自動回復mq，如果設置為false要通過編程實現回復* 3、callback，消費方法，當消費者接收到消息要執行的方法*/channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);}
}

消費者二：sms

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer04_topics_sms {//隊列名稱private static final String QUEUE_INFORM_SMS = "queue_inform_sms";private static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";private static final String ROUTINGKEY_SMS="inform.#.sms.#";public static void main(String[] args) throws IOException, TimeoutException {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");//建立新連接Connection connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成Channel channel = connection.createChannel();/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//聲明一個交換機//參數：String exchange, String type/*** 參數明細：* 1、交換機的名稱* 2、交換機的類型* fanout：對應的rabbitmq的工作模式是 publish/subscribe* direct：對應的Routing	工作模式* topic：對應的Topics工作模式* headers： 對應的headers工作模式*/channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);//進行交換機和隊列綁定//參數：String queue, String exchange, String routingKey/*** 參數明細：* 1、queue 隊列名稱* 2、exchange 交換機名稱* 3、routingKey 路由key，作用是交換機根據路由key的值將消息轉發到指定的隊列中，在發布訂閱模式中調協為空字符串*/channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_TOPICS_INFORM,ROUTINGKEY_SMS);//實現消費方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 當接收到消息后此方法將被調用* @param consumerTag  消費者標簽，用來標識消費者的，在監聽隊列時設置channel.basicConsume* @param envelope 信封，通過envelope* @param properties 消息屬性* @param body 消息內容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交換機String exchange = envelope.getExchange();//消息id，mq在channel中用來標識消息的id，可用于確認消息已接收long deliveryTag = envelope.getDeliveryTag();//消息內容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//監聽隊列//參數：String queue, boolean autoAck, Consumer callback/*** 參數明細：* 1、queue 隊列名稱* 2、autoAck 自動回復，當消費者接收到消息后要告訴mq消息已接收，如果將此參數設置為tru表示會自動回復mq，如果設置為false要通過編程實現回復* 3、callback，消費方法，當消費者接收到消息要執行的方法*/channel.basicConsume(QUEUE_INFORM_SMS,true,defaultConsumer);}
}

3.測試

使用生產者發送若干條消息，交換機根據routingkey統配符匹配并轉發消息到指定的隊列。

4.思考

1、本案例的需求使用Routing工作模式能否實現？
使用Routing模式也可以實現本案例，共設置三個 routingkey，分別是email、sms、all，email隊列綁定email和all，sms隊列綁定sms和all，這樣就可以實現上邊案例的功能，實現過程比topics復雜。
Topic模式更多加強大，它可以實現Routing、publish/subscirbe模式的功能。

五.Header模式

header模式與routing不同的地方在于，header模式取消routingkey，使用header中的 key/value（鍵值對）匹配隊列。
案例：
根據用戶的通知設置去通知用戶，設置接收Email的用戶只接收Email，設置接收sms的用戶只接收sms，設置兩種
通知類型都接收的則兩種通知都有效。

1.生產者

隊列與交換機綁定的代碼與之前不同，如下：

Map<String, Object> headers_email = new Hashtable<String, Object>();
headers_email.put("inform_type", "email");
Map<String, Object> headers_sms = new Hashtable<String, Object>();
headers_sms.put("inform_type", "sms");
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_HEADERS_INFORM,"",headers_sms);

String message = "email inform to user"+i;
Map<String,Object> headers = new Hashtable<String, Object>();
headers.put("inform_type", "email");//匹配email通知消費者綁定的header
//headers.put("inform_type", "sms");//匹配sms通知消費者綁定的header
AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties.Builder();
properties.headers(headers);
//Email通知
channel.basicPublish(EXCHANGE_HEADERS_INFORM, "", properties.build(), message.getBytes());

2.消費者

channel.exchangeDeclare(EXCHANGE_HEADERS_INFORM, BuiltinExchangeType.HEADERS);
Map<String, Object> headers_email = new Hashtable<String, Object>();
headers_email.put("inform_email", "email");
//交換機和隊列綁定
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);
//指定消費隊列
channel.basicConsume(QUEUE_INFORM_EMAIL, true, consumer);

六.RPC

RPC即客戶端遠程調用服務端的方法 ，使用MQ可以實現RPC的異步調用，基于Direct交換機實現，流程如下：
1、客戶端即是生產者就是消費者，向RPC請求隊列發送RPC調用消息，同時監聽RPC響應隊列。
2、服務端監聽RPC請求隊列的消息，收到消息后執行服務端的方法，得到方法返回的結果
3、服務端將RPC方法的結果發送到RPC響應隊列
4、客戶端（RPC調用方）監聽RPC響應隊列，接收到RPC調用結果。