在前面的兩篇部落格中

遇到的例項都是一個訊息只發送給一個消費者（工作者），他們的訊息模型分別為（P代表生產者，C代表消費者，紅色代表隊列）：

這次我們來看下將一個訊息傳送給多個消費者（工作者），這種模式一般被稱為“釋出/訂閱”模式。其工作模型為（P代表生產者，X代表Exchange(路由器/交換機)，C代表消費者，紅色代表隊列）：

我們發現，工作模型中首次出現路由器，並且每個消費者有單獨的佇列。生產者生成訊息後將其傳送給路由器，然後路由器轉送到佇列，消費者各自到自己的佇列裡面獲取訊息進行消費。在實際的應用場景中，生產者一般不會直接將訊息傳送給佇列，而是傳送給路由器進行中轉，Exchange必須清楚的知道怎麼處理收到的訊息：是將訊息傳送到一個特定佇列還是多有佇列，或者直接廢棄訊息。這種才符合RabbitMQ訊息模型的核心思想

接下來我們詳細展開今天的話題：

一、Exchange

Exchange在我們的工作模型中首次出現，因此需要詳細介紹下。

Exchange分為4種類型：

Direct：完全根據key進行投遞的，例如，繫結時設定了routing key為”abc”，那麼客戶端提交的訊息，只有設定了key為”abc”的才會投遞到佇列。
Topic：對key進行模式匹配後進行投遞，符號”#”匹配一個或多個詞，符號”*”匹配正好一個詞。例如”abc.#”匹配”abc.def.ghi”，”abc.*”只匹配”abc.def”。
Fanout：不需要key，它採取廣播模式，一個訊息進來時，投遞到與該交換機繫結的所有佇列。
Headers:我們可以不考慮它。
 

今天我們的例項採用fanout型別的exchange。

儘管首次出現，但是其實我們前面的案例中也有用到exchange，只是我們沒有給他名字，用的是RabbitMQ預設的，比如下面這段程式碼，我們將路由器名這個引數傳入了“”，如果我們需要自己宣告exchange的話，這個就不能傳入“”了，而是傳入自己定義好的值。

二、臨時佇列

前面兩篇部落格中，我們都在使用佇列的時候給出了定義好的名字，這在生產者和消費者共用相同佇列的時候很有必要，但是我們有了exchange，生產者不需要知道有哪些佇列，因此佇列名字可以不用指定了，而是通過RabbitMQ 介面自己去生成臨時佇列，佇列名字也由RabbitMQ自動生成。通過

可以宣告一個非持久的、通道獨佔的、自動刪除的佇列，getQueue()方法可以獲取隨機佇列名字。這個名字用來在佇列和exchange之間建立binding關係的時候使用：

三、程式碼實現

基於上面exchange和臨時佇列的知識鋪墊，可以展開今天的程式碼實現了。

1.  生產者

   public class Product {
       //exchange名字
       public static String EXCHANGE_NAME = "exchange";
   
       public static void main(String[] args) {
           ConnectionFactory factory = new ConnectionFactory();
           factory.setHost("localhost");
           Connection connection = null;
           Channel channel = null;
           try {
               // 1.建立連線和通道
               connection = factory.newConnection();
               channel = connection.createChannel();
   
               // 2.為通道宣告exchange和exchange的型別
               channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
               
               String msg = " hello rabbitmq, this is publish/subscribe mode";
               // 3.傳送訊息到指定的exchange,佇列指定為空,由exchange根據情況判斷需要傳送到哪些佇列
               channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());
               System.out.println("product send a msg: " + msg);
           } catch (IOException e) {
               e.printStackTrace();
           } catch (TimeoutException e) {
               e.printStackTrace();
           } finally {
               // 4.關閉連線
               if (channel != null) {
                   try {
                       channel.close();
                   } catch (IOException e) {
                       e.printStackTrace();
                   } catch (TimeoutException e) {
                       e.printStackTrace();
                   }
               }
   
               if (connection != null) {
                   try {
                       connection.close();
                   } catch (IOException e) {
                       e.printStackTrace();
                   }
               }
           }
       }
   }

2. 消費者1

   public class Consumer1 {
   
       public static void main(String[] args) {
           ConnectionFactory factory = new ConnectionFactory();
           factory.setHost("localhost");
           Connection connection = null;
           Channel channel = null;
           try {
               // 1.建立連線和通道
               connection = factory.newConnection();
               channel = connection.createChannel();
   
               // 2.為通道宣告exchange以及exchange型別
               channel.exchangeDeclare(Product.EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
   
               // 3.建立隨機名字的佇列
               String queueName = channel.queueDeclare().getQueue();
   
               // 4.建立exchange和佇列的繫結關係
               channel.queueBind(queueName, Product.EXCHANGE_NAME, "");
               System.out.println(" **** Consumer1 keep alive ,waiting for messages, and then deal them");
               // 5.通過回撥生成消費者並進行監聽
               Consumer consumer = new DefaultConsumer(channel) {
                   @Override
                   public void handleDelivery(String consumerTag, Envelope envelope,
                           com.rabbitmq.client.AMQP.BasicProperties properties, byte[] body) throws IOException {
   
                       // 獲取訊息內容然後處理
                       String msg = new String(body, "UTF-8");
                       System.out.println("*********** Consumer1" + " get message :[" + msg + "]");
                   }
               };
               // 6.消費訊息
               channel.basicConsume(queueName, true, consumer);
           } catch (IOException e) {
               e.printStackTrace();
           } catch (TimeoutException e) {
               e.printStackTrace();
           }
       }
   }

3. 消費者2，核心程式碼同消費者1一樣，只是在日誌列印上將"Consumer1"改為"Consumer2"而已。這裡不再列出具體程式碼。
4. 先執行消費者1和2，然後執行生產者，觀察控制檯log列印情況：

   生產者：
   product send a msg:  hello rabbitmq, this is publish/subscribe mode
   
   消費者1：
    **** Consumer1 keep alive ,waiting for messages, and then deal them
   *********** Consumer1 get message :[ hello rabbitmq, this is publish/subscribe mode]
   
   消費者2：
    **** Consumer2 keep alive ,waiting for messages, and then deal them
   *********** Consumer2 get message :[ hello rabbitmq, this is publish/subscribe mode]

   可以看到，當生產者發出訊息後，兩個消費者最終都收到了訊息。

5. 我們去檢視RabbitMQ管理頁面：

   在Exchanges 標籤頁裡面多了一個名為“exchange”的路由器，他的型別是fanout。點exchange 的link進入詳細頁面：

   發現在binding專案中有了兩條繫結關係，佇列的名字也可以看到。將頁面切換到Queues標籤頁：

   出現了兩個新的佇列，佇列名字和繫結關係中的一樣，並且佇列都是自動刪除的、通道獨佔的。

6. 然後將消費者1和消費者2都停掉，重新檢視管理頁面，我們發現exchange還在，binding關係不存在了，臨時佇列也自動刪除了