RabbitMQ 訊息佇列 - fanout 模式分發訊息
阿新 • • 發佈:2018-11-16
fanout 模式
將同一個 message 傳送到所有同該 Exchange 繫結的 queue, 只要 RoutingKey 是一樣, 這條訊息都會被投遞到所有與此 Exchange 繫結的 queue 中
p_fanout.php
<?php
//配置資訊
$conn_args = array(
'host' => '127.0.0.1',
'port' => '5672',
'login' => 'admin' ,
'password' => 'admin',
'vhost'=>'/'
);
$exchangeName = 'test_fanout'; //交換機名
$routeKey = 'key_fanout'; //路由key
//建立連線和channel
$conn = new AMQPConnection($conn_args);
$conn->connect() or die("Cannot connect to the broker!\n");
$channel = new AMQPChannel($conn);
//建立交換機物件
$exchange = new AMQPExchange($channel );
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_FANOUT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();
//傳送訊息
for($i=0; $i<5; ++$i){
//訊息內容
$message = "TEST MESSAGE! ".$i;
echo "Send Message:".$exchange->publish($message, $routeKey)."\n" ;
}
$conn->disconnect();
?>
c_fanout_1.php
<?php
//配置資訊
$conn_args = array(
'host' => '127.0.0.1',
'port' => '5672',
'login' => 'admin',
'password' => 'admin',
'vhost'=>'/'
);
$exchangeName = 'test_fanout'; //交換機名
$queueName = 'test_fanout_1'; //佇列名
$routeKey = 'key_fanout'; //路由key
//建立連線和channel
$conn = new AMQPConnection($conn_args);
$conn->connect() or die("Cannot connect to the broker!\n");
$channel = new AMQPChannel($conn);
//建立交換機物件
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_FANOUT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();
$queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();
$queue->bind($exchangeName, $routeKey);
echo "Message:\n";
while(True){
$queue->consume('processMessage');
//自動ACK應答
//$queue->consume('processMessage', AMQP_AUTOACK);
}
$conn->disconnect();
/*
* 消費回撥函式
* 處理訊息
*/
function processMessage($envelope, $q) {
$msg = $envelope->getBody();
echo $msg."\n"; //處理訊息
$q->ack($envelope->getDeliveryTag()); //手動傳送ACK應答
}
?>
c_fanout_2.php
<?php
//配置資訊
$conn_args = array(
'host' => '127.0.0.1',
'port' => '5672',
'login' => 'admin',
'password' => 'admin',
'vhost'=>'/'
);
$exchangeName = 'test_fanout'; //交換機名
$queueName = 'test_fanout_2'; //佇列名
$routeKey = 'key_fanout'; //路由key
//建立連線和channel
$conn = new AMQPConnection($conn_args);
$conn->connect() or die("Cannot connect to the broker!\n");
$channel = new AMQPChannel($conn);
//建立交換機物件
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_FANOUT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();
$queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();
$queue->bind($exchangeName, $routeKey);
echo "Message:\n";
while(True){
$queue->consume('processMessage');
//自動ACK應答
//$queue->consume('processMessage', AMQP_AUTOACK);
}
$conn->disconnect();
/*
* 消費回撥函式
* 處理訊息
*/
function processMessage($envelope, $q) {
$msg = $envelope->getBody();
echo $msg."\n"; //處理訊息
$q->ack($envelope->getDeliveryTag()); //手動傳送ACK應答
}
?>
分別執行 c_fanout_1.php
c_fanout_2.php
等待消費訊息, 然後執行 p_fanout.php
生產訊息
# php c_fanout_1.php
Message:
TEST MESSAGE! 0
TEST MESSAGE! 1
TEST MESSAGE! 2
TEST MESSAGE! 3
TEST MESSAGE! 4
# php c_fanout_2.php
Message:
TEST MESSAGE! 0
TEST MESSAGE! 1
TEST MESSAGE! 2
TEST MESSAGE! 3
TEST MESSAGE! 4
# php p_fanout.php
Send Message:1
Send Message:1
Send Message:1
Send Message:1
Send Message:1