(使用Java客戶端)

一、概述

在前面的專題學習中，我們建立了一個工作佇列，在工作佇列中假如每個任務交給一個確定的工作者，不管是生產者還是消費者都必須知道一個指定的佇列名稱才能傳送和接收訊息，而RabbitMQ訊息模型的核心思想就是生產者不會將訊息直接傳送給佇列。

因為生產者通常不會知道訊息將會被哪些消費者接收，生產者的訊息雖然不是直接傳送給queue(佇列)，但是訊息會交給exchange(交換機)，所以需要定義exchange的訊息分發模式來實現訊息的分發，這便是這部分專題學習中我們將要學習的釋出者/訂閱者模式，這樣實現了訊息生產者和訊息消費者之間的解耦。

在前面的專題學習中實現簡單訊息傳遞和工作佇列中有如下一行程式碼：

　　channel.basicPublish("", queueName, null, msg.getBytes());

在上述程式碼中第一個是空字串其實就是exchangeName,這裡用了空字串，就表示訊息會交給預設的exchange。

為了說明這種訊息分發模型，我們將構建一個簡單的日誌記錄系統，它包括兩個程式--第一個程式用來發送日誌訊息，第二個程式用來接收列印這些日誌訊息。

在日誌記錄系統執行的每個接收者都將接收到訊息，這樣我們可以執行一個接收者將訊息輸出到控制檯。

總的原則：傳送的日誌訊息將被廣播到所有的接收者。

二、日誌訊息系統的實現

2.1、exchange(交換機)

之前傳送和接收訊息都是通過一個佇列來實現，現在是時候介紹下一個完整的RabbitMQ的訊息傳遞模型了。

首先來對之前學習的訊息傳遞加深一下映象

>一個生產者是一個用於傳送訊息的應用

>一個佇列是儲存訊息的緩衝區

>一個消費者是一個接收訊息的應用。

在前面已經提到了RabbitMQ的核心思想是：生產者從來不需要直接傳送任何訊息到佇列中，實際上通常生產者甚至不知道訊息江北傳送到任何一個佇列中。

相反，生產者只能傳送訊息到一個交換元件中(exchange)，exchange是一個很簡單的東西，一方面它接收來自生產者的訊息，另外一方面它將把來自生產者的訊息放入到佇列中，exchange必須知道怎麼接收一個訊息，而且接收的訊息應該被新增到一個指定的佇列？還是多個佇列中，或者接收的訊息被丟棄，這個規則被exchange所定義，它的結構如下：

圖-1

exchange有如下幾種定義型別：direct、topic、headers、fanout，每種型別都自己的實現方式和訊息分發機制，在此我們將重點放在最後一種型別：fanout,首先建立一個這種型別的交換。

channel.exchangeDeclare("logs", "fanout");

基於fanout的exchange是非常簡單的，正如它的名字一樣，我們能猜到它的具體實現，它不僅僅廣播各種來著生產者的訊息到它所知道的所有佇列中，這正是日誌記錄系統的所需要的。

2.2、交換列表(exchange list)

為了列出伺服器中所有的exchanges(交換機)，我們通過執行rabbitmqctl來實現，在列出的列表中有一些amp.*changes和沒有定義名稱的exchange(預設)，這些是被伺服器預設建立的，但是這些當我們需要使用的時候是不可用的。

在之前不知道關於exchange的任何東西，但是它仍然能夠傳送訊息到佇列，這可能因為是使用了預設的exchange，因為我們定義一個空的串("")。

之前釋出的訊息：

channel.basicPublish("", "hello", null, message.getBytes());

第一個引數就是exchange的名稱，空的字串表示預設或者是無名的exchange，訊息被路由到指定的routingKey名稱的隊名，加入它存在的話。

2.3、臨時佇列(Temporary queues)

在之前我們使用的佇列都是被定義過特殊的名稱(hello和task_queue)，對於RabbitMQ來說命名一個佇列是至關重要的，當你想在生產者和消費者中分享佇列的時候，給一個佇列的名稱是必須的。

但是那些都不是日誌記錄系統所需要的，我們希望能夠獲得所有的日誌資訊，而不只是其中的一部分，而且我們只對當前正在傳遞的資訊感興趣，對舊的日誌資訊不感興趣，要解決這些問題，我們需要分兩個步驟：

首先當我們連結到RabbitMQ伺服器的時候，需要一個新的、空的佇列，為了做到這點，可以建立一個隨機名的佇列，或者更好的方法就是讓伺服器選擇一個隨機的佇列名。

其次，當斷開與佇列的連線時，消費者應該被自動刪除掉。

在Java客戶端，我們通過一個無引數的queueDeclare()方法為我們建立一個非持久的、唯一的、能自動刪除的佇列與佇列名稱

String queueName = channel.queueDeclare().getQueue();

在這一點上queueName包含一個隨機佇列名稱，比如它可能看起來像amq.gen-JzTY20BRgKO-HjmUJj0wLg.的隨機串。

2.4、繫結(bindings)

圖-2

我們已經建立了一個fanout exchange和一個佇列，現在我們需要告訴exchange去傳送訊息到佇列中，exchange和佇列之間的關係被稱為一個繫結(binding)。

channel.queueBind(queueName, "logs", "");

從現在開始我們從logs exchange將被新增訊息到佇列中，使用rabbitmqctl list_bingdins能列出所有的繫結。

2.5、釋出者/訂閱者實現(putting it all together)

圖-3

生產者程式碼和之前的傳送訊息的程式碼並沒有太大的區別，最重要的變化是，我們現在要將釋出的訊息傳遞給logs exchange來代替無名的exchange(之前的是"")，在傳送訊息時需要提供一個routingKey，它對於fanout exchange是非常重要的，不能被忽視的，這裡的EmitLog.java程式碼如下：

傳送

package com.xuz.ps;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class EmitLog {
	private static final String EXCHANGE_NAME = "logs";
	
	public static void main(String[] args) throws IOException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("127.0.0.1");
		Connection conn = factory.newConnection();
		Channel channel = conn.createChannel();
		/**exchange型別
		 * direct(直接)、topic(主題)、headers(標題)和fanout
		 */
		channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
		String message = getMessage(args);
		channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
		System.out.println("Sent["+message+"]");
		channel.close();
		conn.close();
	}
	private static String getMessage(String[] strings) {
		if(strings.length<1){
			return "info:Hello World!";
		}
		return joinStrings(strings,"");
	}
	private static String joinStrings(String[] strings, String string) {
		int len = strings.length;
		if(len == 0)return "";
		StringBuilder words = new StringBuilder(strings[0]);
		for (int i = 0; i < len; i++) {
			words.append(string).append(strings[i]);
		}
		return words.toString();
	}
}

接收

package com.xuz.ps;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;
public class ReceiveLogs {
	private static final String EXCHANGE_NAME = "logs";
	
	public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("127.0.0.1");
		Connection conn = factory.newConnection();
		Channel channel = conn.createChannel();
		channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
		//獲取佇列名稱
		String queueName = channel.queueDeclare().getQueue();
		//繫結佇列與exchange
		channel.queueBind(queueName, EXCHANGE_NAME, "");
		System.out.println("ReceiveLogs wait for message .TO exit press CTRL+C");
		
		QueueingConsumer consumer = new QueueingConsumer(channel);
		channel.basicConsume(queueName, true,consumer);
		while(true){
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
		    String message = new String(delivery.getBody());
		    System.out.println("Received [" + message + "]");  
		}
	}
}

2.6、測試釋出者/訂閱者

操作步驟：

1、執行多個ReceiveLogs，分別記為01、02、03、04，首先執行前三個接收者，如下圖所示

     

圖-4

圖-5

圖-6

2、執行EmitLog.java，此時可以看到上述三個接收者都能接收訊息

圖-7

3、執行ReceiveLogs04，此時它沒有收到訊息。

圖-8

4、再次執行EmitLog.java，此時可以看到所有的接收者都接收到了訊息。

圖-9

說明exchange在接收到生產者的訊息後，會將訊息傳送到當前已經與它綁定了的所有的queue中，在接收者完訊息之後，RabbitMQ將佇列中的訊息移除。

原始碼下載