（二）原理 · php&golang

vhosts ：虛擬主機。一個RabbitMQ的實體上可以有多個vhosts，用戶與權限設置就是依附于vhosts。對一般PHP應用，不需要用戶權限設定，直接使用默認就存在的”/”就可以了，用戶可以使用默認就存在的”guest”。一個簡單的配置示例： $conn_args = array( 'host' => '127.0.0.1', 'port' => '5672', 'login' => 'guest', 'password' => 'guest', 'vhost'=>'/' ); connection 與 channel ：連接與信道 connection是指物理的連接，一個client與一個server之間有一個連接；一個連接上可以建立多個channel，可以理解為邏輯上的連接。一般應用的情況下，有一個channel就夠用了，不需要創建更多的channel。示例代碼： //創建連接和channel $conn = new AMQPConnection($conn_args); if (!$conn->connect()) { die("Cannot connect to the broker!\n"); } $channel = new AMQPChannel($conn); exchange 與 routingkey ：交換機與路由鍵為了將不同類型的消息進行區分，設置了交換機與路由兩個概念。比如，將A類型的消息發送到名為‘C1’的交換機，將類型為B的發送到’C2’的交換機。當客戶端連接C1處理隊列消息時，取到的就只是A類型消息。進一步的，如果A類型消息也非常多，需要進一步細化區分，比如某個客戶端只處理A類型消息中針對K用戶的消息，routingkey就是來做這個用途的。 $e_name = 'e_linvo'; //交換機名 $k_route = array(0=> 'key_1', 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->declare()."\n"; for($i=0; $i<5; ++$i){ echo "Send Message:".$ex->publish($message . date('H:i:s'), $k_route[i%2])."\n"; } 對于交換機，有兩個重要的概念： A、類型。有三種類型： Fanout類型最簡單，這種模型忽略routingkey； Direct類型是使用最多的，使用確定的routingkey。這種模型下，接收消息時綁定’key_1’則只接收key_1的消息；最后一種是Topic，這種模式與Direct類似，但是支持通配符進行匹配，比如： ‘key_*’，就會接受key_1和key_2。Topic貌似美好，但是有可能導致不嚴謹，所以還是推薦使用Direct。 B、持久化。指定了持久化的交換機，在重新啟動時才能重建，否則需要客戶端重新聲明生成才行。需要特別明確的概念：交換機的持久化，并不等于消息的持久化。只有在持久化隊列中的消息，才能持久化；如果沒有隊列，消息是沒有地方存儲的；消息本身在投遞時也有一個持久化標志的，PHP中默認投遞到持久化交換機就是持久的消息，不用特別指定。 $e_name = 'e_linvo'; //交換機名 $q_name = 'q_linvo'; //隊列名 $k_route = ''; //路由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 '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"; //處理消息 } 從上述示例中可以看到，交換機既可以由消息發送端創建，也可以由消息消費者創建。創建一個隊列(line:20)后，需要將隊列綁定到交換機上(line:25)隊列才能工作，routingkey也是在這里指定的。有的資料上寫成bindingkey，其實一回事兒，弄兩個名詞反倒容易混淆。消息的處理，是有兩種方式： A，一次性。用 $q->get([…])，不管取到取不到消息都會立即返回，一般情況下使用輪詢處理消息隊列就要用這種方式； B，阻塞。用 $q->consum( callback, […] ) 程序會進入持續偵聽狀態，每收到一個消息就會調用callback指定的函數一次，直到某個callback函數返回FALSE才結束。流控機制當消息生產的速度更快,而進程的處理能力低時,消息就會堆積起來,占用內存越來越多,導致MQ崩潰,所以rabbitmq有一個流控機制,當超過限定時候就會阻止接受消息,mq流控有三種機制 1,主動阻塞住發消息太快的連接,這個無法調整,如果被阻塞了,在abbitmqctl 控制臺上會顯示一個blocked的狀態。 2,內存超過限量,會阻塞連接,在vm_memory_high_watermark可調 3,剩余磁盤在限定以下mq會主動阻塞所有的生產者,默認為50m,在disk_free_limit可調.