Rabbitmq學習(一) Rabbitmq初探

Rabbitmq學習(一) Rabbitmq初探

理論
定義
消息隊列：在消息的傳輸過程中保存消息的的容器。

這是一個較為經典的消費-生產者模型，說起來比較抽象，打個比方：A線程需要給B線程發送消息（A、B線程不一定是在同一臺機器上的），A線程先把消息發送到消息隊列服務器上，然后B線程去讀取或是訂閱消息服務器上消息隊列中的消息，線程A和B之間并沒有進行直接通信。MQ服務器在中間起到中繼的作用。

適用的應用場景
比較適合異步傳輸，這里解釋一下什么是異步和同步。

異步：發送方不關心消息有沒有發送成功，只發送消息，不去獲取消息是否發送成功。

同步：發送方關心消息是否發送成功，發送消息后，會等待接收方返回狀態碼，根據狀態碼來判斷是否發送成功，然后執行相對于的動作。

下邊以Http中的同步和異步為例：

如：普通的B/S架構客戶端和服務器端之間的通信就是同步的，即提交請求 ---> 等待服務器處理完畢返回消息 ---> 拿到服務器返回的消息，處理完畢。

如：Ajax技術就是異步的，請求通過事件觸發 ---> 服務器處理（瀏覽器不用等待，仍可以做其他的事情） ---> 處理完畢。

有人可能會好奇說應用場景怎么說到了同步和異步，那說明你還不是很理解技術和應用場景之間的緊密聯系。

工作過程

生產者客戶端：
客戶端連接到RabbitMQ服務器上，打開一個消息通道（channel）；
客戶端聲明一個消息交換機（exchange），并設置相關屬性。
客戶端聲明一個消息隊列（queue），并設置相關屬性。
客戶端使用routing key在消息交換機（exchange）和消息隊列（queue）中建立好綁定關系。
客戶端投遞消息都消息交換機（exchange）上
客戶端關閉消息通道（channel）以及和服務器的連接。
服務器端：
exchange接收到消息后，根據消息的key（這個key的產生規則暫時沒研究，有知道的小伙伴可以留言告訴我）和以及設置的binding，進行消息路由，將消息投遞到一個或多個消息隊列中。

安裝
由于rabbitMq需要erlang語言的支持，在安裝rabbitMq之前需要安裝erlang，執行命令：

sudo apt-get install erlang

這樣就安裝完了。
接下來，安裝rabbitMq：
sudo apt-get install rabbitmq-server

安裝完之后啟動rabbitMQ

創建用戶
sudo rabbitmqctl add_user lsl 123456

將用戶設置為管理員（只有管理員才能遠程登錄）

sudo rabbitmqctl set_user_tags 用戶名 administrator

同時為用戶設置讀寫等權限
sudo rabbitmqctl set_permissions -p / lsl ".*" ".*" ".*"

跟著訪問 http://192.168.136.130:15672/
就能看到配置頁面了，這也說明已經成功安裝rabbitmq

實戰消費者和生產者（PHP版本）
生產者：生產消息，發送消息。類似工廠。
消費者：接受消息，使用消息。類似顧客。
隊列：存儲消息。類似倉庫、中轉站。隊列可以存儲很多的消息，因為它基本上是一個無限制的緩沖區，前提是你的機器有足夠的存儲空間。多個生產者可以將消息發送到同一個隊列中，多個消費者也可以只從同一個隊列接收數據。這就是隊列的特性。

下面寫一個demo來實現rabbitmq的消費者和生產者
send.php

<?php
//配置信息 
$conn_args = array(
    'host' => '127.0.0.1',
    'port' => '5672',
    'login' => 'lsl',
    'password' => '123456',
    'vhost'=>'/'
);
$e_name = 'e_linvo'; //交換機名 
//$q_name = 'q_linvo'; //無需隊列名 
$k_route = 'key_1'; //路由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);
date_default_timezone_set("Asia/Shanghai");
//發送消息 
//$channel->startTransaction(); //開始事務  
for($i=0; $i<5; ++$i){
    sleep(2);//每個兩秒發送一條消息
    //消息內容 
    $message = "HelloWorld!".date("h:i:sa");
    echo "Send Message:".$ex->publish($message, $k_route)."\n";
}
//$channel->commitTransaction(); //提交事務 

$conn->disconnect();
?>

rec.php

<?php
//配置信息 
$conn_args = array(
    'host' => '127.0.0.1',
    'port' => '5672',
    'login' => 'lsl',
    'password' => '123456',
    'vhost'=>'/'
);
$e_name = 'e_linvo'; //交換機名 
$q_name = 'q_linvo'; //隊列名 
$k_route = 'key_1'; //路由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 "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應答  
}
$conn->disconnect();

先讓消費者接收消息

再調用生產者發送消息

再查看消費者收到的消息

假設消費者掛掉了，看看消息是怎么樣的

看一下之前部署的，你會發現linvo這個隊列里有5條消息，這意味著沒有消費者去讀取它，把消息堆積在隊列里了

當你打開消費者，頁面的Total值就馬上變為0了，這意味著消息已經被接收。
這樣就模擬了隊列對消息的處理。