原理流程
生產(chǎn)者主要做的是：創(chuàng)建連接-->創(chuàng)建channel-->創(chuàng)建交換機對象-->發(fā)送消息
消費者主要做的是：創(chuàng)建連接-->創(chuàng)建channel-->創(chuàng)建交換機-->創(chuàng)建隊列-->綁定交換機/隊列/路由鍵-->接收消息
案例一：
生產(chǎn)者
<?php
$conn_args = array(
'host' => '127.0.0.1',
'port' => '5672',
'login' => 'admin',
'password' => 'admin',
'vhost'=>'/'
);
$content = isset($_GET['content']) ? htmlspecialchars($_GET['content']) : '請輸入消息';
//創(chuàng)建連接和channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($conn);
//創(chuàng)建交換機
$e_name = 'e_linvo'; //交換機名
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct類型
$ex->setFlags(AMQP_DURABLE); //持久化
$ex->declare();
//echo "Send Message:".$ex->publish("TEST MESSAGE，key_1 by xust" . date('H:i:s', time()), 'key_1')."\n";
//echo "Send Message:".$ex->publish("TEST MESSAGE，key_2 by xust" . date('H:i:s', time()), 'key_2')."\n";
$ex->publish($content, 'key_1');
echo "finish\n";
消費者
<?php
$conn_args = array(
'host' => '127.0.0.1',
'port' => '5672',
'login' => 'admin',
'password' => 'admin',
'vhost'=>'/'
);
$e_name = 'e_linvo'; //交換機名
$q_name = 'q_linvo'; //隊列名
$k_route = 'key_2'; //路由key
//創(chuàng)建連接和channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($conn);
//創(chuàng)建交換機
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct類型
$ex->setFlags(AMQP_DURABLE); //持久化
$ex->declare();
//創(chuàng)建隊列
$q = new AMQPQueue($channel);
$q->setName($q_name);
$q->setFlags(AMQP_DURABLE); //持久化
$q->declare(); //最好隊列object在這里declare()下蜈亩，否則如果是新的queue會報錯
//綁定交換機與隊列，并指定路由鍵弃甥，可以多個路由鍵
$q->bind($e_name, 'key_1');
//$q->bind($e_name, 'key_33');
//阻塞模式接收消息
echo "Message:\n";
$q->consume('processMessage', AMQP_AUTOACK); //自動ACK應(yīng)答
$conn->disconnect();
/**
* 消費回調(diào)函數(shù)
* 處理消息
*/
function processMessage($envelope, $queue) {
var_dump($envelope->getRoutingKey());
$msg = $envelope->getBody();
mysql_insert($msg);
}
//消息入庫
function mysql_insert($json){
$pdo = new PDO("mysql:host=localhost;dbname=rabbitmq","root","000000",array(PDO::ATTR_PERSISTENT => true) );
$sql = "insert into rabbitmq_table(sJson) values ('{$json}')";
echo $sql."\n";
if ($pdo->exec($sql)){
echo "insert success\n";
} else {
echo "error\n";
}
}
表結(jié)構(gòu)
DROP TABLE IF EXISTS `rabbitmq_table`;
CREATE TABLE `rabbitmq_table` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`sJson` varchar(255) NOT NULL DEFAULT '',
PRIMARY KEY (`id`)
) ENGINE=MyISAM DEFAULT CHARSET=utf8;
啟動消費者程序缝左，同時不斷請求生產(chǎn)者程序当娱，查看數(shù)據(jù)變化
41.png
案例二：
生產(chǎn)者
<?php
$routingkey='key2';
//設(shè)置你的連接
$conn_args = array('host' => 'localhost', 'port' => '5672', 'login' => 'admin', 'password' => 'admin');
$conn = new AMQPConnection($conn_args);
if ($conn->connect()) {
echo "鏈接成功\n";
}else {
echo "鏈接失敗 \n ";
}
//你的消息
$message = json_encode(array('消息測試','My RabbitMQ-PHP','消息隊列'));
//創(chuàng)建channel
$channel = new AMQPChannel($conn);
//創(chuàng)建exchange
$ex = new AMQPExchange($channel);
$ex->setName('exchange');//創(chuàng)建名字
$ex->setType(AMQP_EX_TYPE_DIRECT);
$ex->setFlags(AMQP_DURABLE);
//$ex->setFlags(AMQP_AUTODELETE);
//echo "exchange status:".$ex->declare();
echo "exchange status:".$ex->declareExchange();
echo "\n";
for($i=0;$i<10;$i++){
if($routingkey=='key2'){
$routingkey='key';
}else{
$routingkey='key2';
}
$ex->publish($message,$routingkey);
}
消費者
<?php
$bindingkey='key2';
//連接RabbitMQ
$conn_args = array( 'host'=>'127.0.0.1' , 'port'=> '5672', 'login'=>'admin' , 'password'=> 'admin','vhost' =>'/');
$conn = new AMQPConnection($conn_args);
$conn->connect();
//設(shè)置queue名稱豺裆，使用exchange拒秘，綁定routingkey
$channel = new AMQPChannel($conn);
$q = new AMQPQueue($channel);
$q->setName('queue2');
$q->setFlags(AMQP_DURABLE);
$q->declare();
$q->bind('exchange',$bindingkey);
//消息獲取
$messages = $q->get(AMQP_AUTOACK) ;
if ($messages){
var_dump(json_decode($messages->getBody(), true ));
}
$conn->disconnect();
案例三：
生產(chǎn)者
<?php
$conn_args = array(
'host' => '127.0.0.1',
'port' => '5672',
'login' => 'admin',
'password' => 'admin',
'vhost'=>'/'
);
$e_name = 'e_linvo'; //交換機名
//$q_name = 'q_linvo'; //無需隊列名
$k_route = 'key_1'; //路由key
//創(chuàng)建連接和channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($conn);
//消息內(nèi)容
$message = "我的RabbitMQ走通了~~";
//創(chuàng)建交換機對象
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
//發(fā)送消息
//$channel->startTransaction(); //開始事務(wù)
for($i=0; $i<5; ++$i){
echo "Send Message:".$ex->publish($message, $k_route)."\n";
}
//$channel->commitTransaction(); //提交事務(wù)
$conn->disconnect();
消費者
<?php
//配置信息
$conn_args = array(
'host' => '127.0.0.1',
'port' => '5672',
'login' => 'admin',
'password' => 'admin',
'vhost'=>'/'
);
$e_name = 'e_linvo'; //交換機名
$q_name = 'q_linvo'; //隊列名
$k_route = 'key_1'; //路由key
//創(chuàng)建連接和channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($conn);
//創(chuàng)建交換機
$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";
//創(chuàng)建隊列
$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應(yīng)答
}
$conn->disconnect();
/**
* 消費回調(diào)函數(shù)
* 處理消息
*/
function processMessage($envelope, $queue) {
$msg = $envelope->getBody();
echo $msg."\n"; //處理消息
$queue->ack($envelope->getDeliveryTag()); //手動發(fā)送ACK應(yīng)答
}
然后分別命令行執(zhí)行兩個程序臭猜，消費者程序處于監(jiān)聽狀態(tài)躺酒，多次執(zhí)行生產(chǎn)者測試，觀察效果
php /data/wwwroot/default/publisher.php //執(zhí)行生產(chǎn)者
php /data/wwwroot/default/consumer.php //執(zhí)行消費者
42.png
附加刪除隊列：
43.png
