## 工作過程 1. 創建一個vhost. 2. 在該vhost上創建一個connection. 3. 在該connection上創建一個channel. 4. 在該channel上創建一個exchange(如果有就跳過,前提是屬性一致).需要設置的屬性有交換機名稱,類型,是否持久化,是否rabbitmq內部使用,默認false,擴展參數. 5. 在該channel上創建一個queue.需要設置的屬性有隊列名稱,,是否持久化,是否自動刪除, 6. 將exchange和queue通過路由鍵綁定(binding key). 7. 發布消息. ## vhost 一個RabbitMQ的實體上可以有多個vhosts,用戶與權限的設置就是依賴于vhosts.對于一般PHP應用,不需要用戶權限設定,直接使用默認就存在的的 "/" 就可以了,用戶可以使用默認就存在的"guest"賬戶 .一個簡單的配置示例 ``` $config = [ 'host' => '127.0.0.1', 'port' => '5672', 'login' => 'guest', 'password' => 'guest', 'vhost' => '/' ]; ``` ## conenction 與 channel conenction是值物理的連接,一個client與一個server之間有一個連接,一個連接上可以建立多個channel,可以理解為邏輯上的連接.一般應用的情況下,有一個 channel就夠用了,不需要創建更多的channel. ``` //創建連接和channel $conn = new AMQPConnection($config); if (!$conn->connect()) { die("cannot connect to the broker\n"); } $channel = new AMQPChannel($conn); ``` ## exchaneg和routingKey 為了將不同類型的消息進行區分,設置了 交換機與路由兩個概念,如果,將A類型的消息發送到名為"C1"的交換機,將類型為B的消息發送到"C2"的交換機.當客戶端連接C1處理隊列消息時,渠道的就只是A類型消息.進一步的,如果A類型消息也非常多,需要進一步細化分區,比如某個客戶端只處理A類型消息中針對K用于的消息,routingKey就是用來做這個用途的. ~~~ $e_name = 'elinvo'; //交換機名 $k_route = ['key_1', 'key_2']; //路由key //創建交換機 $ex = new AMQPExchange($channel); $ex->setName($e_name); $ex->setType(AMQP_EX_TYPE_DIRECT); //direct類型 $ex->setFlags(AMQP_DURABLE); //持久化 echo "Exchange Status: {$ex->declareExchange()} \n"; for ($i = 0; $i < 5; $i++) { echo 'Send Message : ' . $ex->publish($message) . date('H:i:s'), $k_route[i % 2] . "\n"; } ~~~ 從上面代碼可以看到,發送消息時,只要有"交換機"就夠了,至于交換機后面有沒有對應的處理隊列,發送發是不用管的.routkingKey可以是空的字符串.在示例中,我們使用了兩個可以交替發送消息,是為了下面更便于理解routingKey的作用. 對于交換機,有兩個重要的概念: 1. 類型,有三種類型: * Fanout類型最簡單,這種模型忽略routingKey. * Direct類型是使用最大的,使用確定的routingKey.這種模式下,接收消息時綁定"key_1"則只接收"key_1"的消息. * Topic類型與Direct類似,但是支持通配符進行匹配,比如:"key_*"就會接收"key_1"和"key_2".Topic貌似美好,但是有可能導致不嚴謹,所以還是推薦使用Direct. 2. 持久化,指定了持久化的交換機,在重新啟動時才能重建,否則需要客戶端重新聲明生成才行. * 需要特別明確的概念:交換機的持久化,并不等于消息的持久化.只有在持久化隊列中的消息,才能持久化.如果沒有隊列,消息是沒有地方存儲的.消息本身在投遞時也有一個持久化標志的 ,PHP中默認投遞到持久化交換機就是持久的消息,不用特別指定. ## QUEUE 事實上,隊列僅是針對接收方(customer)的,由接收方根據需求創建的.只有隊列創建了,交換機才會將新接收到的消息送到隊列中,交換機是不會將在隊列創建之前的消息放進來的.換句話說,在建立隊列之前,發出的所有消息都被丟棄了.下面這個圖比官方的圖更清楚--QUEUE是屬于ReceiveMessage的一部分.  下面看一下創建隊列及接收消息的示例: ~~~ $e_name = 'e_linvo'; //交換機名 $q_name = 'q_linvo'; //隊列名 $k_route = ''; //路由key $conn = new AMQPConnection($config); 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->declareExchange() . "\n"; //創建隊列 $q = new AMQPQueue($channel); $q->setName($q_name); $q->setFlags(AMQP_DURABLE); //持久化 //綁定交換機與隊列,并制定路由key echo "Queue Bind:" . $q->bind($e_name, $k_route) . "\n"; //阻塞模式接收消息 echo "Message :\n"; $q->consume('processMessage', AMQP_AUTOACK); //自動ACK應答 $conn->disconnect(); //消費回調函數,處理消息 function processMessage($envelope, $queue) { var_dump($envelope->getRoutingKey); $msg = $envelope->getBody(); echo $msg . "\n"; } ~~~ 從上述示例可以看到,交換機即可以由消息發送端創建,也可以由消息消費者創建. 創建一個隊列后,需要將隊列綁定到交換機上,隊列才能工作,routingKey也是在這里指定的.有的資料寫成bindingKey,其實是一回事. 消息的處理,是有兩種方式: 1. 一次性: 用$q->get([...]),不管渠道娶不到消息都會立即返回,一般情況下使用輪詢處理消息隊列就要用這種方式. 2. 阻塞:用$q->consum(callback,[...]) ,程序會進入持續偵聽狀態,沒收到一個消息就會調用callback指定的函數一次,直到某個callback函數返回false才結束. 關于callback,PHP的是指出數組的,可以使用['對象名','對象方法']的方式傳參. 在上述示例中,使用的routingKey="",意味著接收全部的消息.我們可以將其改為routingKey = "key_1",可以看到結果中僅有設置routingKey為"key_1"的 內容了. 注意:routingKey="key_1"與routingKey = "key_2"是兩個不同的隊列. 假設:celint1與celint2都連接到key_1的隊列上,一個消息被client1處理之后,就不會被client2處理。而 routingkey = " "是另類，client_all綁定到 " " 上，將消息全都處理后，client1和client2上也就沒有消息了. 在程序設計上，需要規劃好exchange的名稱，以及如何使用key區分開不同類型的標記，在消息產生的地方插入發送消息代碼。后端處理，可以針對每一個key啟動一個或多個client，以提高消息處理的實時性。如何使用PHP進行多線程的消息處理，將在下一節中講述.