程序師世界是廣大編程愛好者互助、分享、學習的平台,程序師世界有你更精彩!
首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫知識
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
 程式師世界 >> 編程語言 >> 網頁編程 >> PHP編程 >> 關於PHP編程 >> 用PHP嘗試RabbitMQ(amqp擴展)

用PHP嘗試RabbitMQ(amqp擴展)

編輯:關於PHP編程

兩年前曾發過一篇《用Python嘗試RabbitMQ》,沒想到兩年後的今天,基於PHP的amqp資料還是少得可憐,原來的幾個擴展也都一個個被廢棄,只剩amqp還健在,且被PECL收錄。雖說被收錄,可官方手冊中的信息還是略顯單薄。
言歸正傳,amqp擴展的安裝就不多說了,可參看前幾天發的那篇《給PHP安裝amqp擴展》
裝好了amqp後就可以開始編寫代碼了:

消費者:接收消息

邏輯:
創建連接-->創建channel-->創建交換機-->創建隊列-->綁定交換機/隊列/路由鍵-->接收消息
[php]
<?php   
/*************************************
 * PHP amqp(RabbitMQ) Demo - consumer
 * Author: Linvo
 * Date: 2012/7/30
 *************************************/ 
//配置信息 
$conn_args = array( 
    'host' => '192.168.1.93',  
    '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應答 

生產者:發送消息
邏輯:
創建連接-->創建channel-->創建交換機對象-->發送消息
[php] 
<?php   
/*************************************
 * PHP amqp(RabbitMQ) Demo - publisher
 * Author: Linvo
 * Date: 2012/7/30
 *************************************/ 
//配置信息 
$conn_args = array( 
    'host' => '192.168.1.93',  
    '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);   
 
//消息內容 
$message = "TEST MESSAGE! 測試消息!";   
 
//創建交換機對象    
$ex = new AMQPExchange($channel);   
$ex->setName($e_name);   
 
//發送消息 
//$channel->startTransaction(); //開始事務  
for($i=0; $i<5; ++$i){ 
    echo "Send Message:".$ex->publish($message, $k_route)."\n";  

//$channel->commitTransaction(); //提交事務 
 
$conn->disconnect();   
需要注意的地方是:

queue對象有兩個方法可用於取消息:consume和get。
前者是阻塞的,無消息時會被掛起,適合循環中使用;
後者則是非阻塞的,取消息時有則取,無則返回false。

測試截圖

運行消費者:

 

運行生產者,發消息:


消費者接收到消息:


 


作者:linvo

  1. 上一頁:
  2. 下一頁:
Copyright © 程式師世界 All Rights Reserved