RabbitMQ與PHP（一）－ RabbitMQ的原理與操作範例

RabbitMQ是流行的開源訊息佇列系統，用erlang語言開發，完整的實作了AMQP（高階訊息佇列協定）。網站在： http://www.rabbitmq.com/  上面有教學和實例程式碼（Python和Java的）。
7af40ad162d9f2d36b6bf89fa8ec8a136327cc4c
AMPQ協定為了能夠滿足各種訊息佇列需求，在概念上比較複雜。首先，rabbitMQ啟動預設是沒有任何設定的，需要客戶端連線上去，設定交換器等才能運作。不把這些基礎概念弄清楚，後面程式設計就容易產生問題。
1.vhosts ： 虛擬主機。
一個RabbitMQ的實體上可以有多個vhosts，使用者與權限設定就是依附在vhosts。對一般PHP應用，不需要使用者權限設定，直接使用預設就存在的"/"就可以了，使用者可以使用預設就存在的"guest"。一個簡單的設定範例：
$conn_args = array(
    'host' => '127.0.0.1',
    'port' => '5672',
 o. 'guest',
    'vhost'=>'/'
);
2.connection 與channel ：  連接與通道
connection是指物理的連接，一個client與一個server之間有一個連接；一個連接上可以建立多個channel，可以理解為邏輯上的連結。一般應用的情況下，有一個channel就夠用了，不需要創造更多的channel。範例程式碼：
//建立連線和channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
    die("Cannot connect to the broker!n");
}
$channel = new AMQPChannel($conn);
3.exchange 與  routingkey ： 交換器與路由鍵
為了將不同類型的訊息區分，設定了交換器與路由兩個概念。例如，將A類型的消息傳送到名為‘C1’的交換機，將類型為B的發送到'C2'的交換機。當客戶端連接C1處理佇列訊息時，取到的就只是A類型訊息。進一步的，如果A類型訊息也非常多，需要進一步細化區分，例如某個客戶端只處理A類型訊息中針對K用戶的訊息，routingkey就是來做這個用途的。
$e_name = 'e_linvo'; //交換器名稱
$k_route = array(0=> 'key_1', 1=> 'key_2'); //路由key
//建立交換器
$ex = new AMQPExchange($ channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct類型
$ex->setFlags(AMQP_DURABLE); //持久化
echo "Exchange:". $ex->declare()."n";
for($i=0; $i    echo "Send Message:".$ex->publish($message . date( 'H:i:s'), $k_route[i%2])."n";
}
由以上代碼可以看到，發送訊息時，只要有「交換器」就夠了。至於交換器後面有沒有對應的處理佇列，發送方是不用管的。 routingkey可以是空的字串。在範例中，我使用了兩個key交替發送訊息，是為了下面更方便理解routingkey的作用。
對於交換機，有兩個重要的概念：
A，類型。有三種類型： Fanout類型最簡單，這種模型忽略routingkey；Direct類型是使用最多的，使用確定的routingkey。這種模型下，接收訊息時綁定'key_1'則只接收key_1的訊息；最後一種是Topic，這種模式與Direct類似，但是支援通配符進行匹配，例如： 'key_*'，就會接受key_1和key_2。 Topic看起來像美好，但有可能導致不嚴謹，所以還是推薦使用Direct。
B，持久化。指定了持久化的交換機，在重新啟動時才能重建，否則需要客戶端重新聲明產生才行。
需要特別明確的概念：交換器的持久化，並不等於訊息的持久化。只有在持久化佇列中的消息，才能持久化；如果沒有佇列，訊息是沒有地方儲存的；訊息本身在投遞時也有一個持久化標誌的，PHP中預設投遞到持久化交換器就是持久的訊息，不用特別指定。
4.queue： 隊列
講了這麼多，才講到隊列呀。事實上，佇列僅是針對接收方（consumer）的，由接收者根據需求創建的。只有佇列建立了，交換器才會將新接受到的訊息送到佇列中，交換器是不會在佇列建立之前的訊息放進來的。換句話說，在建立佇列之前，發出的所有訊息都被丟棄了。下面這個圖比RabbitMQ官方的圖更清楚──Queue是屬於ReceiveMessage的一部分。
024f78f0f736afc37053e415b219ebc4b7451266
接下來看一下建立佇列及接收訊息的範例：
$e_name = 'e_linvo'; //交換器名稱編號
//建立連線與channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
    die("Cannot connect to the broker!n");
}
$channel = new AMQPChannel($conn);
//建立交換器
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT); //directECT);類型
$ex->setFlags(AMQP_DURABLE); //持久化
echo "Exchange Status:".$ex->declare()."n";
//建立佇列
$q = new AMQPQueue($channel) ;
$q->setName($q_name);
$q->setFlags(AMQP_DURABLE); //持久化
//綁定交換器與佇列，並指定路由鍵
echo 'Queue Bind: '.$q- >bind($e_name, $k_route)."n";
//阻塞模式接收訊息
echo "Message:n";
$q->consume('processMessage', AMQP_AUTOACK); //自動答應$
ACK$ conn->disconnect();
/**
* 消費回呼函數
* 處理訊息
*/
function processMessage($envelope, $queue) {
    var_dump($envelope->getRoutingKey);
  echo $msg."n"; //處理訊息
}
從上述範例可以看到，交換器既可以由訊息傳送端創建，也可以由訊息消費者建立。
建立一個佇列(line:20)後，需要將佇列綁定到交換器上(line:25)佇列才能運作，routingkey也是在這裡指定的。有的資料上寫成bindingkey，其實一回事兒，弄兩個名詞反倒容易混淆。
訊息的處理，有兩種方式：
A，一次性。用 $q->get([...])，不管拿不到訊息都會立即返回，一般情況下使用輪詢處理訊息佇列就要用這種方式；
B，阻塞。用 $q->consum( callback, [...] ) 程式會進入持續偵聽狀態，每收到一個訊息就會呼叫callback指定的函數一次，直到某個callback函數回傳FALSE才結束。
關於callback，這裡多說幾句： PHP的call_back是支援使用陣列的，例如： $c = new MyClass(); $c->counter = 100; $q->consume( array($c,'myfunc ') ) 這樣就可以呼叫自己寫的處理類別。 MyClass中myfunc的參數定義，與上例中processMessage一樣就行。
在上述範例中，使用的$routingkey = ''， 意味著接收全部的訊息。我們可以將其改為 $routingkey = 'key_1'，可以看到結果中僅有設定routingkey為key_1的內容了。
注意： routingkey = 'key_1'  與 routingkey = 'key_2' 是兩個不同的佇列。假設： client1 與 client2 都連接到 key_1 的佇列上，一個訊息被client1處理之後，就不會被client2處理。而 routingkey = '' 是另類，client_all綁定到 '' 上，將訊息全都處理後，client1和client2上也就沒有消息了。
在程式設計上，需要規劃好exchange的名稱，以及如何使用key區分開不同類型的標記，在訊息產生的地方插入發送訊息代碼。後端處理，可以針對每一個key啟動一個或多個client，以提高訊息處理的即時性。如何使用PHP進行多執行緒的訊息處理，將在下一節中講述。
更多訊息模型，可參考：http://www.rabbitmq.com/tutorials/tutorial-two-python.html
b03533fa828ba61e15fc0e5f4034970a304e59b4
http://nonfu.
以上就介紹了RabbitMQ與PHP（一）－ RabbitMQ的原理與操作範例，包括了方面的內容，希望對PHP教學有興趣的朋友有所幫助。