文章目錄

• 2_RabbitMQ工作模式
• • 1.Work queues
  • 2.Publish/Subscribe
  • • 1.工作模式
    • 2.代碼
    • • 1.生產(chǎn)者
      • • 1.指定消息隊(duì)列相關(guān)消息
        • 2.建立連接&綁定隊(duì)列
        • 3.發(fā)送消息
        • 完整代碼：
      • 2.消費(fèi)者
      • • 1.指定消息隊(duì)列相關(guān)消息
        • 2.建立連接&綁定隊(duì)列
        • 3.實(shí)現(xiàn)消費(fèi)方法&監(jiān)聽(tīng)消息
        • 完整代碼
    • 3.小結(jié)
  • 3.Routing
  • • 1.工作模式
    • 2.代碼
    • • 1.生產(chǎn)者
      • 2.消費(fèi)者
    • 3.小結(jié)
  • 4.Topics
  • • 1.工作模式
    • 2.代碼
    • • 1.生產(chǎn)者
      • 2.消費(fèi)者
    • 3.小結(jié)
  • 5.Header&Rpc
  • • header模式
    • • 1）生產(chǎn)者
      • 2）發(fā)送郵件消費(fèi)者
    • Rpc

2_RabbitMQ工作模式

前置知識(shí)：工作模式指的是某條消息該怎么分發(fā)到隊(duì)列？分發(fā)策略

1.Work queues

work queues與入門程序相比，多了一個(gè)消費(fèi)端，兩個(gè)消費(fèi)端共同消費(fèi)同一個(gè)隊(duì)列中的消息。

應(yīng)用場(chǎng)景：對(duì)于 任務(wù)過(guò)重或任務(wù)較多情況使用工作隊(duì)列可以提高任務(wù)處理的速度。

測(cè)試：

1、使用入門程序，啟動(dòng)多個(gè)消費(fèi)者。

2、生產(chǎn)者發(fā)送多個(gè)消息。

結(jié)果：

1、一條消息只會(huì)被一個(gè)消費(fèi)者接收；

2、rabbit采用輪詢的方式將消息是平均發(fā)送給消費(fèi)者的；

3、消費(fèi)者在處理完某條消息后，才會(huì)收到下一條消息。

2.Publish/Subscribe

1.工作模式

發(fā)布訂閱模式：

1、每個(gè)消費(fèi)者監(jiān)聽(tīng)自己的隊(duì)列。

2、生產(chǎn)者將消息發(fā)給broker，由交換機(jī)將消息轉(zhuǎn)發(fā)到綁定此交換機(jī)的每個(gè)隊(duì)列，每個(gè)綁定交換機(jī)的隊(duì)列都將接收到消息

此交換機(jī)的概念類似計(jì)算機(jī)網(wǎng)絡(luò)中的交換機(jī)，接受包然后分發(fā)下去

2.代碼

案例：

用戶通知，當(dāng)用戶充值成功或轉(zhuǎn)賬完成系統(tǒng)通知用戶，通知方式有短信、郵件多種方法 。

1.生產(chǎn)者

聲明exchange_routing_inform交換機(jī)。

聲明兩個(gè)隊(duì)列并且綁定到此交換機(jī)，綁定時(shí)需要指定routingkey

發(fā)送消息時(shí)需要指定routingkey

1.指定消息隊(duì)列相關(guān)消息

//隊(duì)列名稱 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";//通過(guò)連接工廠創(chuàng)建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設(shè)置虛擬機(jī)，一個(gè)mq服務(wù)可以設(shè)置多個(gè)虛擬機(jī)，每個(gè)虛擬機(jī)就相當(dāng)于一個(gè)獨(dú)立的mqconnectionFactory.setVirtualHost("/");

2.建立連接&綁定隊(duì)列

Connection connection = null; Channel channel = null;//建立新連接 connection = connectionFactory.newConnection(); //創(chuàng)建會(huì)話通道,生產(chǎn)者和mq服務(wù)所有通信都在channel通道中完成 channel = connection.createChannel(); //聲明隊(duì)列，如果隊(duì)列在mq 中沒(méi)有則要?jiǎng)?chuàng)建 //參數(shù)：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments /*** 參數(shù)明細(xì)* 1、queue 隊(duì)列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊(duì)列還在* 3、exclusive 是否獨(dú)占連接，隊(duì)列只允許在該連接中訪問(wèn)，如果connection連接關(guān)閉隊(duì)列則自動(dòng)刪除,如果將此參數(shù)設(shè)置true可用于臨時(shí)隊(duì)列的創(chuàng)建* 4、autoDelete 自動(dòng)刪除，隊(duì)列不再使用時(shí)是否自動(dòng)刪除此隊(duì)列，如果將此參數(shù)和exclusive參數(shù)設(shè)置為true就可以實(shí)現(xiàn)臨時(shí)隊(duì)列（隊(duì)列不用了就自動(dòng)刪除）* 5、arguments 參數(shù)，可以設(shè)置一個(gè)隊(duì)列的擴(kuò)展參數(shù)，比如：可設(shè)置存活時(shí)間*/ channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null); channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null); //聲明一個(gè)交換機(jī) //參數(shù)：String exchange, String type /*** 參數(shù)明細(xì)：* 1、交換機(jī)的名稱* 2、交換機(jī)的類型* fanout：對(duì)應(yīng)的rabbitmq的工作模式是 publish/subscribe* direct：對(duì)應(yīng)的Routing 工作模式* topic：對(duì)應(yīng)的Topics工作模式* headers： 對(duì)應(yīng)的headers工作模式*/ channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT); //進(jìn)行交換機(jī)和隊(duì)列綁定 //參數(shù)：String queue, String exchange, String routingKey /*** 參數(shù)明細(xì)：* 1、queue 隊(duì)列名稱* 2、exchange 交換機(jī)名稱* 3、routingKey 路由key，作用是交換機(jī)根據(jù)路由key的值將消息轉(zhuǎn)發(fā)到指定的隊(duì)列中，在發(fā)布訂閱模式中調(diào)協(xié)為空字符串*/ channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,""); channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");

3.發(fā)送消息

//發(fā)送消息 //參數(shù)：String exchange, String routingKey, BasicProperties props, byte[] body /*** 參數(shù)明細(xì)：* 1、exchange，交換機(jī)，如果不指定將使用mq的默認(rèn)交換機(jī)（設(shè)置為""）* 2、routingKey，路由key，交換機(jī)根據(jù)路由key來(lái)將消息轉(zhuǎn)發(fā)到指定的隊(duì)列，如果使用默認(rèn)交換機(jī)，routingKey設(shè)置為隊(duì)列的名稱* 3、props，消息的屬性* 4、body，消息內(nèi)容*/ for(int i=0;i<20;i++){//消息內(nèi)容String message = "send inform message to user "+(i+1);channel.basicPublish(EXCHANGE_FANOUT_INFORM,"",null,message.getBytes());System.out.println("send to mq "+message); }

完整代碼：

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;/*** @author Administrator* @version 1.0* @create 2018-06-17 18:10**/ public class Producer02_publish {//隊(duì)列名稱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) {//通過(guò)連接工廠創(chuàng)建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設(shè)置虛擬機(jī)，一個(gè)mq服務(wù)可以設(shè)置多個(gè)虛擬機(jī)，每個(gè)虛擬機(jī)就相當(dāng)于一個(gè)獨(dú)立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新連接connection = connectionFactory.newConnection();//創(chuàng)建會(huì)話通道,生產(chǎn)者和mq服務(wù)所有通信都在channel通道中完成channel = connection.createChannel();//聲明隊(duì)列，如果隊(duì)列在mq 中沒(méi)有則要?jiǎng)?chuàng)建//參數(shù)：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 參數(shù)明細(xì)* 1、queue 隊(duì)列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊(duì)列還在* 3、exclusive 是否獨(dú)占連接，隊(duì)列只允許在該連接中訪問(wèn)，如果connection連接關(guān)閉隊(duì)列則自動(dòng)刪除,如果將此參數(shù)設(shè)置true可用于臨時(shí)隊(duì)列的創(chuàng)建* 4、autoDelete 自動(dòng)刪除，隊(duì)列不再使用時(shí)是否自動(dòng)刪除此隊(duì)列，如果將此參數(shù)和exclusive參數(shù)設(shè)置為true就可以實(shí)現(xiàn)臨時(shí)隊(duì)列（隊(duì)列不用了就自動(dòng)刪除）* 5、arguments 參數(shù)，可以設(shè)置一個(gè)隊(duì)列的擴(kuò)展參數(shù)，比如：可設(shè)置存活時(shí)間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//聲明一個(gè)交換機(jī)//參數(shù)：String exchange, String type/*** 參數(shù)明細(xì)：* 1、交換機(jī)的名稱* 2、交換機(jī)的類型* fanout：對(duì)應(yīng)的rabbitmq的工作模式是 publish/subscribe* direct：對(duì)應(yīng)的Routing 工作模式* topic：對(duì)應(yīng)的Topics工作模式* headers： 對(duì)應(yīng)的headers工作模式*/channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);//進(jìn)行交換機(jī)和隊(duì)列綁定//參數(shù)：String queue, String exchange, String routingKey/*** 參數(shù)明細(xì)：* 1、queue 隊(duì)列名稱* 2、exchange 交換機(jī)名稱* 3、routingKey 路由key，作用是交換機(jī)根據(jù)路由key的值將消息轉(zhuǎn)發(fā)到指定的隊(duì)列中，在發(fā)布訂閱模式中調(diào)協(xié)為空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");//發(fā)送消息//參數(shù)：String exchange, String routingKey, BasicProperties props, byte[] body/*** 參數(shù)明細(xì)：* 1、exchange，交換機(jī)，如果不指定將使用mq的默認(rèn)交換機(jī)（設(shè)置為""）* 2、routingKey，路由key，交換機(jī)根據(jù)路由key來(lái)將消息轉(zhuǎn)發(fā)到指定的隊(duì)列，如果使用默認(rèn)交換機(jī)，routingKey設(shè)置為隊(duì)列的名稱* 3、props，消息的屬性* 4、body，消息內(nèi)容*/for(int i=0;i<20;i++){//消息內(nèi)容String message = "send inform message to user "+(i+1);channel.basicPublish(EXCHANGE_FANOUT_INFORM,"",null,message.getBytes());System.out.println("send to mq "+message);}} catch (Exception e) {e.printStackTrace();} finally {//關(guān)閉連接//先關(guān)閉通道try {channel.close();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}try {connection.close();} catch (IOException e) {e.printStackTrace();}}} }

2.消費(fèi)者

1.指定消息隊(duì)列相關(guān)消息

//隊(duì)列名稱 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";//通過(guò)連接工廠創(chuàng)建新的連接和mq建立連接 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("127.0.0.1"); connectionFactory.setPort(5672);//端口 connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); //設(shè)置虛擬機(jī)，一個(gè)mq服務(wù)可以設(shè)置多個(gè)虛擬機(jī)，每個(gè)虛擬機(jī)就相當(dāng)于一個(gè)獨(dú)立的mq connectionFactory.setVirtualHost("/");

2.建立連接&綁定隊(duì)列

Connection connection = null; Channel channel = null;//建立新連接 connection = connectionFactory.newConnection(); //創(chuàng)建會(huì)話通道,生產(chǎn)者和mq服務(wù)所有通信都在channel通道中完成 channel = connection.createChannel(); //聲明隊(duì)列，如果隊(duì)列在mq 中沒(méi)有則要?jiǎng)?chuàng)建 //參數(shù)：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments /*** 參數(shù)明細(xì)* 1、queue 隊(duì)列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊(duì)列還在* 3、exclusive 是否獨(dú)占連接，隊(duì)列只允許在該連接中訪問(wèn)，如果connection連接關(guān)閉隊(duì)列則自動(dòng)刪除,如果將此參數(shù)設(shè)置true可用于臨時(shí)隊(duì)列的創(chuàng)建* 4、autoDelete 自動(dòng)刪除，隊(duì)列不再使用時(shí)是否自動(dòng)刪除此隊(duì)列，如果將此參數(shù)和exclusive參數(shù)設(shè)置為true就可以實(shí)現(xiàn)臨時(shí)隊(duì)列（隊(duì)列不用了就自動(dòng)刪除）* 5、arguments 參數(shù)，可以設(shè)置一個(gè)隊(duì)列的擴(kuò)展參數(shù)，比如：可設(shè)置存活時(shí)間*/ channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null); channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null); //聲明一個(gè)交換機(jī) //參數(shù)：String exchange, String type /*** 參數(shù)明細(xì)：* 1、交換機(jī)的名稱* 2、交換機(jī)的類型* fanout：對(duì)應(yīng)的rabbitmq的工作模式是 publish/subscribe* direct：對(duì)應(yīng)的Routing 工作模式* topic：對(duì)應(yīng)的Topics工作模式* headers： 對(duì)應(yīng)的headers工作模式*/ channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT); //進(jìn)行交換機(jī)和隊(duì)列綁定 //參數(shù)：String queue, String exchange, String routingKey /*** 參數(shù)明細(xì)：* 1、queue 隊(duì)列名稱* 2、exchange 交換機(jī)名稱* 3、routingKey 路由key，作用是交換機(jī)根據(jù)路由key的值將消息轉(zhuǎn)發(fā)到指定的隊(duì)列中，在發(fā)布訂閱模式中調(diào)協(xié)為空字符串*/ channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,""); channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");

3.實(shí)現(xiàn)消費(fèi)方法&監(jiān)聽(tīng)消息

//實(shí)現(xiàn)消費(fèi)方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 當(dāng)接收到消息后此方法將被調(diào)用* @param consumerTag 消費(fèi)者標(biāo)簽，用來(lái)標(biāo)識(shí)消費(fèi)者的，在監(jiān)聽(tīng)隊(duì)列時(shí)設(shè)置channel.basicConsume* @param envelope 信封，通過(guò)envelope* @param properties 消息屬性* @param body 消息內(nèi)容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交換機(jī)String exchange = envelope.getExchange();//消息id，mq在channel中用來(lái)標(biāo)識(shí)消息的id，可用于確認(rèn)消息已接收long deliveryTag = envelope.getDeliveryTag();//消息內(nèi)容String message= new String(body,"utf-8");System.out.println("receive message:"+message);} }; //監(jiān)聽(tīng)隊(duì)列 //參數(shù)：String queue, boolean autoAck, Consumer callback /*** 參數(shù)明細(xì)：* 1、queue 隊(duì)列名稱* 2、autoAck 自動(dòng)回復(fù)，當(dāng)消費(fèi)者接收到消息后要告訴mq消息已接收，如果將此參數(shù)設(shè)置為tru表示會(huì)自動(dòng)回復(fù)mq，如果設(shè)置為false要通過(guò)編程實(shí)現(xiàn)回復(fù)* 3、callback，消費(fèi)方法，當(dāng)消費(fèi)者接收到消息要執(zhí)行的方法*/ 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;/*** @author Administrator* @version 1.0* @create 2018-06-17 18:22**/ public class Consumer02_subscribe_email {//隊(duì)列名稱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 {//通過(guò)連接工廠創(chuàng)建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設(shè)置虛擬機(jī)，一個(gè)mq服務(wù)可以設(shè)置多個(gè)虛擬機(jī)，每個(gè)虛擬機(jī)就相當(dāng)于一個(gè)獨(dú)立的mqconnectionFactory.setVirtualHost("/");//建立新連接Connection connection = connectionFactory.newConnection();//創(chuàng)建會(huì)話通道,生產(chǎn)者和mq服務(wù)所有通信都在channel通道中完成Channel channel = connection.createChannel();/*** 參數(shù)明細(xì)* 1、queue 隊(duì)列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊(duì)列還在* 3、exclusive 是否獨(dú)占連接，隊(duì)列只允許在該連接中訪問(wèn)，如果connection連接關(guān)閉隊(duì)列則自動(dòng)刪除,如果將此參數(shù)設(shè)置true可用于臨時(shí)隊(duì)列的創(chuàng)建* 4、autoDelete 自動(dòng)刪除，隊(duì)列不再使用時(shí)是否自動(dòng)刪除此隊(duì)列，如果將此參數(shù)和exclusive參數(shù)設(shè)置為true就可以實(shí)現(xiàn)臨時(shí)隊(duì)列（隊(duì)列不用了就自動(dòng)刪除）* 5、arguments 參數(shù)，可以設(shè)置一個(gè)隊(duì)列的擴(kuò)展參數(shù)，比如：可設(shè)置存活時(shí)間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);//聲明一個(gè)交換機(jī)//參數(shù)：String exchange, String type/*** 參數(shù)明細(xì)：* 1、交換機(jī)的名稱* 2、交換機(jī)的類型* fanout：對(duì)應(yīng)的rabbitmq的工作模式是 publish/subscribe* direct：對(duì)應(yīng)的Routing 工作模式* topic：對(duì)應(yīng)的Topics工作模式* headers： 對(duì)應(yīng)的headers工作模式*/channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);//進(jìn)行交換機(jī)和隊(duì)列綁定//參數(shù)：String queue, String exchange, String routingKey/*** 參數(shù)明細(xì)：* 1、queue 隊(duì)列名稱* 2、exchange 交換機(jī)名稱* 3、routingKey 路由key，作用是交換機(jī)根據(jù)路由key的值將消息轉(zhuǎn)發(fā)到指定的隊(duì)列中，在發(fā)布訂閱模式中調(diào)協(xié)為空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");//實(shí)現(xiàn)消費(fèi)方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 當(dāng)接收到消息后此方法將被調(diào)用* @param consumerTag 消費(fèi)者標(biāo)簽，用來(lái)標(biāo)識(shí)消費(fèi)者的，在監(jiān)聽(tīng)隊(duì)列時(shí)設(shè)置channel.basicConsume* @param envelope 信封，通過(guò)envelope* @param properties 消息屬性* @param body 消息內(nèi)容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交換機(jī)String exchange = envelope.getExchange();//消息id，mq在channel中用來(lái)標(biāo)識(shí)消息的id，可用于確認(rèn)消息已接收long deliveryTag = envelope.getDeliveryTag();//消息內(nèi)容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//監(jiān)聽(tīng)隊(duì)列//參數(shù)：String queue, boolean autoAck, Consumer callback/*** 參數(shù)明細(xì)：* 1、queue 隊(duì)列名稱* 2、autoAck 自動(dòng)回復(fù)，當(dāng)消費(fèi)者接收到消息后要告訴mq消息已接收，如果將此參數(shù)設(shè)置為tru表示會(huì)自動(dòng)回復(fù)mq，如果設(shè)置為false要通過(guò)編程實(shí)現(xiàn)回復(fù)* 3、callback，消費(fèi)方法，當(dāng)消費(fèi)者接收到消息要執(zhí)行的方法*/channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);} }

3.小結(jié)

3.Routing

1.工作模式

路由模式：

1、每個(gè)消費(fèi)者監(jiān)聽(tīng)自己的隊(duì)列，并且設(shè)置routingkey。

2、生產(chǎn)者將消息發(fā)給交換機(jī)，由交換機(jī)根據(jù)routingkey來(lái)轉(zhuǎn)發(fā)消息到指定的隊(duì)列。

2.代碼

1.生產(chǎn)者

全部代碼：

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;/*** @author Administrator* @version 1.0* @create 2018-06-17 19:23**/ public class Producer03_routing {//隊(duì)列名稱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) {//通過(guò)連接工廠創(chuàng)建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設(shè)置虛擬機(jī)，一個(gè)mq服務(wù)可以設(shè)置多個(gè)虛擬機(jī)，每個(gè)虛擬機(jī)就相當(dāng)于一個(gè)獨(dú)立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新連接connection = connectionFactory.newConnection();//創(chuàng)建會(huì)話通道,生產(chǎn)者和mq服務(wù)所有通信都在channel通道中完成channel = connection.createChannel();//聲明隊(duì)列，如果隊(duì)列在mq 中沒(méi)有則要?jiǎng)?chuàng)建//參數(shù)：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 參數(shù)明細(xì)* 1、queue 隊(duì)列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊(duì)列還在* 3、exclusive 是否獨(dú)占連接，隊(duì)列只允許在該連接中訪問(wèn)，如果connection連接關(guān)閉隊(duì)列則自動(dòng)刪除,如果將此參數(shù)設(shè)置true可用于臨時(shí)隊(duì)列的創(chuàng)建* 4、autoDelete 自動(dòng)刪除，隊(duì)列不再使用時(shí)是否自動(dòng)刪除此隊(duì)列，如果將此參數(shù)和exclusive參數(shù)設(shè)置為true就可以實(shí)現(xiàn)臨時(shí)隊(duì)列（隊(duì)列不用了就自動(dòng)刪除）* 5、arguments 參數(shù)，可以設(shè)置一個(gè)隊(duì)列的擴(kuò)展參數(shù)，比如：可設(shè)置存活時(shí)間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//聲明一個(gè)交換機(jī)//參數(shù)：String exchange, String type/*** 參數(shù)明細(xì)：* 1、交換機(jī)的名稱* 2、交換機(jī)的類型* fanout：對(duì)應(yīng)的rabbitmq的工作模式是 publish/subscribe* direct：對(duì)應(yīng)的Routing 工作模式* topic：對(duì)應(yīng)的Topics工作模式* headers： 對(duì)應(yīng)的headers工作模式*/channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);//進(jìn)行交換機(jī)和隊(duì)列綁定//參數(shù)：String queue, String exchange, String routingKey/*** 參數(shù)明細(xì)：* 1、queue 隊(duì)列名稱* 2、exchange 交換機(jī)名稱* 3、routingKey 路由key，作用是交換機(jī)根據(jù)路由key的值將消息轉(zhuǎn)發(fā)到指定的隊(duì)列中，在發(fā)布訂閱模式中調(diào)協(xié)為空字符串*/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");//發(fā)送消息//參數(shù)：String exchange, String routingKey, BasicProperties props, byte[] body/*** 參數(shù)明細(xì)：* 1、exchange，交換機(jī)，如果不指定將使用mq的默認(rèn)交換機(jī)（設(shè)置為""）* 2、routingKey，路由key，交換機(jī)根據(jù)路由key來(lái)將消息轉(zhuǎn)發(fā)到指定的隊(duì)列，如果使用默認(rèn)交換機(jī)，routingKey設(shè)置為隊(duì)列的名稱* 3、props，消息的屬性* 4、body，消息內(nèi)容*//* for(int i=0;i<5;i++){//發(fā)送消息的時(shí)候指定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++){//發(fā)送消息的時(shí)候指定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++){//發(fā)送消息的時(shí)候指定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 {//關(guān)閉連接//先關(guān)閉通道try {channel.close();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}try {connection.close();} catch (IOException e) {e.printStackTrace();}}} }

2.消費(fèi)者

全部代碼：

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

3.小結(jié)

4.Topics

1.工作模式

2.代碼

1.生產(chǎn)者

聲明exchange_routing_inform交換機(jī)。

聲明兩個(gè)隊(duì)列并且綁定到此交換機(jī)，綁定時(shí)需要指定routingkey

發(fā)送消息時(shí)需要指定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;/*** @author Administrator* @version 1.0* @create 2018-06-17 19:23**/ public class Producer04_topics {//隊(duì)列名稱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) {//通過(guò)連接工廠創(chuàng)建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設(shè)置虛擬機(jī)，一個(gè)mq服務(wù)可以設(shè)置多個(gè)虛擬機(jī)，每個(gè)虛擬機(jī)就相當(dāng)于一個(gè)獨(dú)立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新連接connection = connectionFactory.newConnection();//創(chuàng)建會(huì)話通道,生產(chǎn)者和mq服務(wù)所有通信都在channel通道中完成channel = connection.createChannel();//聲明隊(duì)列，如果隊(duì)列在mq 中沒(méi)有則要?jiǎng)?chuàng)建//參數(shù)：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 參數(shù)明細(xì)* 1、queue 隊(duì)列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊(duì)列還在* 3、exclusive 是否獨(dú)占連接，隊(duì)列只允許在該連接中訪問(wèn)，如果connection連接關(guān)閉隊(duì)列則自動(dòng)刪除,如果將此參數(shù)設(shè)置true可用于臨時(shí)隊(duì)列的創(chuàng)建* 4、autoDelete 自動(dòng)刪除，隊(duì)列不再使用時(shí)是否自動(dòng)刪除此隊(duì)列，如果將此參數(shù)和exclusive參數(shù)設(shè)置為true就可以實(shí)現(xiàn)臨時(shí)隊(duì)列（隊(duì)列不用了就自動(dòng)刪除）* 5、arguments 參數(shù)，可以設(shè)置一個(gè)隊(duì)列的擴(kuò)展參數(shù)，比如：可設(shè)置存活時(shí)間*/channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);//聲明一個(gè)交換機(jī)//參數(shù)：String exchange, String type/*** 參數(shù)明細(xì)：* 1、交換機(jī)的名稱* 2、交換機(jī)的類型* fanout：對(duì)應(yīng)的rabbitmq的工作模式是 publish/subscribe* direct：對(duì)應(yīng)的Routing 工作模式* topic：對(duì)應(yīng)的Topics工作模式* headers： 對(duì)應(yīng)的headers工作模式*/channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);//進(jìn)行交換機(jī)和隊(duì)列綁定//參數(shù)：String queue, String exchange, String routingKey/*** 參數(shù)明細(xì)：* 1、queue 隊(duì)列名稱* 2、exchange 交換機(jī)名稱* 3、routingKey 路由key，作用是交換機(jī)根據(jù)路由key的值將消息轉(zhuǎn)發(fā)到指定的隊(duì)列中，在發(fā)布訂閱模式中調(diào)協(xié)為空字符串*/channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_EMAIL);channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_SMS);//發(fā)送消息//參數(shù)：String exchange, String routingKey, BasicProperties props, byte[] body/*** 參數(shù)明細(xì)：* 1、exchange，交換機(jī)，如果不指定將使用mq的默認(rèn)交換機(jī)（設(shè)置為""）* 2、routingKey，路由key，交換機(jī)根據(jù)路由key來(lái)將消息轉(zhuǎn)發(fā)到指定的隊(duì)列，如果使用默認(rèn)交換機(jī)，routingKey設(shè)置為隊(duì)列的名稱* 3、props，消息的屬性* 4、body，消息內(nèi)容*/for(int i=0;i<5;i++){//發(fā)送消息的時(shí)候指定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++){//發(fā)送消息的時(shí)候指定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++){//發(fā)送消息的時(shí)候指定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 {//關(guān)閉連接//先關(guān)閉通道try {channel.close();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}try {connection.close();} catch (IOException e) {e.printStackTrace();}}} }

2.消費(fèi)者

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

3.小結(jié)

5.Header&Rpc

header模式

header模式與routing不同的地方在于，header模式取消routingkey，使用header中的 key/value（鍵值對(duì)）匹配

隊(duì)列。

案例：

根據(jù)用戶的通知設(shè)置去通知用戶，設(shè)置接收Email的用戶只接收Email，設(shè)置接收sms的用戶只接收sms，設(shè)置兩種

通知類型都接收的則兩種通知都有效。

1）生產(chǎn)者

隊(duì)列與交換機(jī)綁定的代碼與之前不同，如下：

通知：

2）發(fā)送郵件消費(fèi)者

Rpc

RPC即客戶端遠(yuǎn)程調(diào)用服務(wù)端的方法 ，使用MQ可以實(shí)現(xiàn)RPC的異步調(diào)用，基于Direct交換機(jī)實(shí)現(xiàn)，流程如下：

1、客戶端即是生產(chǎn)者就是消費(fèi)者，向RPC請(qǐng)求隊(duì)列發(fā)送RPC調(diào)用消息，同時(shí)監(jiān)聽(tīng)RPC響應(yīng)隊(duì)列。

2、服務(wù)端監(jiān)聽(tīng)RPC請(qǐng)求隊(duì)列的消息，收到消息后執(zhí)行服務(wù)端的方法，得到方法返回的結(jié)果

3、服務(wù)端將RPC方法 的結(jié)果發(fā)送到RPC響應(yīng)隊(duì)列

4、客戶端（RPC調(diào)用方）監(jiān)聽(tīng)RPC響應(yīng)隊(duì)列，接收到RPC調(diào)用結(jié)果。

總結(jié)

以上是生活随笔為你收集整理的2_RabbitMQ工作模式_Work queues_Publish/Subscribe_Routing_Topics_HeaderRpc的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。