[TOC] ## 概述 * 將消息推送到多個不同隊列,如將日志發送給 log,屏幕輸出 * fanout模式模式不像direct模式通過routingkey來進行匹配，而是會把消息發送到所以的已經綁定的隊列中 ### code 發布消息 <details> <summary>fanout.php</summary> ``` $exchangeName = 'log'; $message = 'log--'; $connection = new AMQPConnection(array('host' => '10.99.121.137', 'port' => '5672', 'vhost' => '/', 'login' => 'superrd', 'password' => 'superrd')); $connection->connect() or die("Cannot connect to the broker!\n"); try { $channel = new AMQPChannel($connection); $exchange = new AMQPExchange($channel); $exchange->setName($exchangeName); $exchange->setType(AMQP_EX_TYPE_FANOUT); $exchange->setFlags(AMQP_DURABLE); $exchange->declareExchange(); for($i=0 ; $i<100;$i++){ $exchange->publish($message.$i,""); var_dump("[x] Sent $message $i"); } } catch (AMQPConnectionException $e) { var_dump($e); exit(); } $connection->disconnect(); ``` </details> <br /> 消費者 ca <details> <summary>ca.php</summary> ``` $exchangeName = 'log'; $queueName = 'queuea'; $routeKey = ''; $connection = new AMQPConnection(array('host' => '10.99.121.137','port' => '5672', 'vhost' => '/', 'login' => 'superrd', 'password' => 'superrd')); $connection->connect() or die("Cannot connect to the broker!\n"); $channel = new AMQPChannel($connection); $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'); //$queue->consume('processMessage', AMQP_AUTOACK); //自動ACK應答 } $conn->disconnect(); /** * 消費回調函數 * 處理消息 */ function processMessage($envelope, $q) { $msg = $envelope->getBody(); sleep(1); echo $msg."\n"; //處理消息 $q->ack($envelope->getDeliveryTag()); //手動發送ACK應答 } ``` </details> <br /> 消費者 cb <details> <summary>cb.php</summary> ``` $exchangeName = 'log'; $queueName = 'queueb'; $routeKey = ''; $connection = new AMQPConnection(array('host' => '10.99.121.137','port' => '5672', 'vhost' => '/', 'login' => 'superrd', 'password' => 'superrd')); $connection->connect() or die("Cannot connect to the broker!\n"); $channel = new AMQPChannel($connection); $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'); //$queue->consume('processMessage', AMQP_AUTOACK); //自動ACK應答 } $conn->disconnect(); /** * 消費回調函數 * 處理消息 */ function processMessage($envelope, $q) { $msg = $envelope->getBody(); sleep(1); echo $msg."\n"; //處理消息 $q->ack($envelope->getDeliveryTag()); //手動發送ACK應答 } ``` </details> <br />