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

時間 2019-12-14

標籤 rabbitmq rabbitmq+php php 演示實例欄目 RabbitMQ 简体版

原文原文鏈接

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

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

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

如：註冊用戶這種服務，它可能解耦成好幾種獨立的服務（帳號驗證，郵箱驗證碼，手機短信碼等）。它們做爲消費者，等待用戶輸入數據，在前臺數據提交以後會通過分解併發送到各個服務所在的url，分發的那個角色就至關於生產者。消費者在獲取數據時候有可能一次不能處理完，那麼它們各自有一個請求隊列，那就是內存緩衝區了。作這項工做的框架叫作消息隊列。windows

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

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

即便在下單時庫存系統不能正常使用。也不影響正常下單，由於下單後，訂單系統寫入消息隊列就再也不關心其餘的後續操做了。服務器

給一張結構圖：併發

幾個概念說明：

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應答 
}
?>

View Code

新建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();
?>

View Code

測試一下：

先起一個窗口一樣切換到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