初識RabbitMQ,附RabbitMQ+PHP演示實例

RabbitMQ是一個在AMQP基礎上實現的企業級消息系統。何謂消息系統，就是消息隊列系統，消息隊列是“”消費-生產者模型“”的一個典型的代表，一端往消息隊列中不斷寫入消息，而另一端則可以讀取或者訂閱隊列中的消息。

what？消費-生產者模型？對，沒錯！就是大學操作系統課程里面的“消費者-生產者模式”，記得當時被這個問題坑的不輕啊。

在項目中，將一些無需即時返回且耗時的操作提取出來，進行了異步操作，而這種異步處理的方式大大的節省了服務器的請求時間，從而提高了系統的吞吐量。而且不影響服務器做其他相應，不獨占服務器資源。

如：注冊用戶這種服務，它可能解耦成好幾種獨立的服務（賬號驗證，郵箱驗證碼，手機短信碼等）。它們作為消費者，等待用戶輸入數據，在前台數據提交之后會經過分解並發送到各個服務所在的url，分發的那個角色就相當於生產者。消費者在獲取數據時候有可能一次不能處理完，那么它們各自有一個請求隊列，那就是內存緩沖區了。做這項工作的框架叫做消息隊列。

又比如：電商系統中的訂單處理系統，傳統處理模式是：下訂單的時候，訂單系統可能會調用庫存系統的接口，這樣兩個系統之間存在一個嚴重依賴關系，如果庫存系統宕機，那么整個流程都會受到影響。現在大多公司的處理方法是：引入消息隊列，下完訂單，訂單系統完成持久化處理，將消息寫入消息隊列，返回用戶訂單下單成功。

對庫存系統來說，采用拉/推的方式，獲取下單信息，庫存系統根據下單信息，進行庫存操作。這樣實現了兩個系統間的解耦。

即使在下單時庫存系統不能正常使用。也不影響正常下單，因為下單后，訂單系統寫入消息隊列就不再關心其他的后續操作了。

 給一張結構圖：

幾個概念說明：

Broker：簡單來說就是消息隊列服務器實體。
　　Exchange：消息交換機，它指定消息按什么規則，路由到哪個隊列。
　　Queue：消息隊列載體，每個消息都會被投入到一個或多個隊列。
　　Binding：綁定，它的作用就是把exchange和queue按照路由規則綁定起來。
　　Routing Key：路由關鍵字，exchange根據這個關鍵字進行消息投遞。
　　vhost：虛擬主機，一個broker里可以開設多個vhost，用作不同用戶的權限分離。
　　producer：消息生產者，就是投遞消息的程序。
　　consumer：消息消費者，就是接受消息的程序。
　　channel：消息通道，在客戶端的每個連接里，可建立多個channel，每個channel代表一個會話任務。

消息隊列的使用過程大概如下：

（1）客戶端連接到消息隊列服務器，打開一個channel。
　　（2）客戶端聲明一個exchange，並設置相關屬性。
　　（3）客戶端聲明一個queue，並設置相關屬性。
　　（4）客戶端使用routing key，在exchange和queue之間建立好綁定關系。
　　（5）客戶端投遞消息到exchange。

exchange接收到消息后，就根據消息的key和已經設置的binding，進行消息路由，將消息投遞到一個或多個隊列里。

exchange也有幾個類型，完全根據key進行投遞的叫做Direct交換機，例如，綁定時設置了routing key為”abc”，那么客戶端提交的消息，只有設置了key為”abc”的才會投遞到隊列。對key進行模式匹配后進行投遞的叫做Topic交換機，符號”#”匹配一個或多個詞，符號”*”匹配正好一個詞。例如”abc.#”匹配”abc.def.ghi”，”abc.*”只匹配”abc.def”。還有一種不需要key的，叫做Fanout交換機，它采取廣播模式，一個消息進來時，投遞到與該交換機綁定的所有隊列。

RabbitMQ支持消息的持久化，也就是數據寫在磁盤上，為了數據安全考慮，我想大多數用戶都會選擇持久化。消息隊列持久化包括3個部分：
　　（1）exchange持久化，在聲明時指定durable => 1
　　（2）queue持久化，在聲明時指定durable => 1
　　（3）消息持久化，在投遞時指定delivery_mode => 2（1是非持久化）

如果exchange和queue都是持久化的，那么它們之間的binding也是持久化的。如果exchange和queue兩者之間有一個持久化，一個非持久化，就不允許建立綁定。

 

好了，講了這么多基本講清楚了RabbitMQ的應用場景和好處，下面我們在windows平台上練一把手，更直觀的來看看RabbitMQ到底是什么？

那么我蠻來安裝RabbitMQ+PHP環境：

1.安裝RabbitMQ

 安裝RabbitMQ之前首先要安裝Erlang語言開發包，下載地址：http://www.erlang.org/download/otp_win32_R15B.exe 默認安裝即可

　  配置環境變量 ERLANG_HOME C:\Program Files (x86)\erl5.9 

      添加到PATH  %ERLANG_HOME%\bin;

 下載安裝RabbitMQ，下載地址：http://www.rabbitmq.com/releases/rabbitmq-server/v3.3.4/rabbitmq-server-3.3.4.exe 

      配置環境變量 C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-2.8.0

      添加到PATH %RABBITMQ_SERVER%\sbin;

  然后到dos里面切換到RabbitMQ目錄下，執行rabbitmq-plugins.bat enable rabbitmq_management， 安裝完成之后以管理員身份啟動 rabbitmq：輸入命令：

　　rabbitmq-service.bat stop

　　rabbitmq-service.bat install

　　rabbitmq-service.bat start

然后，瀏覽器中輸入:127.0.0.1:15672,用戶名密碼是guest ,如果能登陸就說明安裝成功。

 

 

2.接下來要安裝php的amqp擴展

先用phpinfo()查看php版本信息，及,信息

最后根據上面的信息去下載相應的amqp版本：http://pecl.php.net/package/amqp

據上面信息我們的是32位非線程安全版本

加壓后：

將php_amqp.dll復制到php/ext，同時在php.ini中添加如下代碼：

[amqp]  

extension=php_amqp.dll 

然后將rabbitmq.1.dll復制到php根目錄C:/xampp/php/，同時修改apache配置文件httpd.conf，添加如下代碼：

# rabbitmq

LoadFile  "C:/xampp/php/rabbitmq.1.dll"  

最后重啟看看是否已經加載了amqp模塊：

 

---------------------------------到這里為止，安裝已經結束-------------------------------------------------

 

RabbitMQ+PHP展示實例

新建rabbit_consumer.php作為消費者

<?php 
//配置信息 
$conn_args = array( 
    'host' => '127.0.0.1',  
    'port' => '5672',  
    'login' => 'guest',  
    'password' => 'guest', 
    'vhost'=>'/' 
);   
$e_name = 'e_linvo'; //交換機名 
$q_name = 'q_linvo'; //隊列名 
$k_route = 'key_1'; //路由key 
 
//創建連接和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); //direct類型  
$ex->setFlags(AMQP_DURABLE); //持久化 
echo "Exchange Status:".$ex->declare()."\n";   
   
//創建隊列    
$q = new AMQPQueue($channel); 
$q->setName($q_name);   
$q->setFlags(AMQP_DURABLE); //持久化  
echo "Message Total:".$q->declare()."\n";   
 
//綁定交換機與隊列，並指定路由鍵 
echo 'Queue Bind: '.$q->bind($e_name, $k_route)."\n"; 
 
//阻塞模式接收消息 
echo "Message:\n";   
while(True){ 
    $q->consume('processMessage');   
    //$q->consume('processMessage', AMQP_AUTOACK); //自動ACK應答  
} 
$conn->disconnect();   
 
/**
 * 消費回調函數
 * 處理消息
 */ 
function processMessage($envelope, $queue) { 
    $msg = $envelope->getBody(); 
    echo $msg."\n"; //處理消息 
    $queue->ack($envelope->getDeliveryTag()); //手動發送ACK應答 
}
?>

 

 新建rabbit_publisher.php作為生產者

<?php
//配置信息 
$conn_args = array( 
    'host' => '127.0.0.1',  
    'port' => '5672',  
    'login' => 'guest',  
    'password' => 'guest', 
    'vhost'=>'/' 
);   
$e_name = 'e_linvo'; //交換機名 
//$q_name = 'q_linvo'; //無需隊列名 
$k_route = 'key_1'; //路由key 
 
//創建連接和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);   
date_default_timezone_set("Asia/Shanghai");
//發送消息 
//$channel->startTransaction(); //開始事務  
for($i=0; $i<5; ++$i){ 
    sleep(1);//休眠1秒
    //消息內容 
    $message = "TEST MESSAGE!".date("h:i:sa");   
    echo "Send Message:".$ex->publish($message, $k_route)."\n";  
} 
//$channel->commitTransaction(); //提交事務 
 
$conn->disconnect();
?>

測試一下：

先起一個窗口同樣切換到php目錄，輸入:php c:/xampp/htdocs/RabbitMQ/rabbit_consumer.php

運行消費者

 

然后再起一個dos窗口，切換到php根目錄，輸入以下命令：php c:/xampp/htdocs/RabbitMQ/rabbit_publisher.php

運行生產者

消費者接收到消息

 

 這樣就模擬了隊列對消息的處理，希望我們通過這篇文章對RabbitMQ的認識都能有一定的提升。

 另外給出官網的php使用指南：http://www.rabbitmq.com/tutorials/tutorial-one-php.html