2. 程式人生 > >rabbitmq - (訊息佇列) 的基本原理介紹

rabbitmq - (訊息佇列) 的基本原理介紹

阿新 • • 發佈：2018-12-20

介紹

MQ全稱為Message Queue, 是一種分散式應用程式的的通訊方法，它是消費-生產者模型的一個典型的代表，producer往訊息佇列中不斷寫入訊息，而另一端consumer則可以讀取或者訂閱佇列中的訊息。RabbitMQ是MQ產品的典型代表，是一款基於AMQP協議可複用的企業訊息系統

系統架構

Rabbitmq系統最核心的元件是Exchange和Queue，Exchange和Queue是在rabbitmq server（又叫做broker）端，producer和consumer在應用端。

原理大致圖(MQ:Message Queue):

Queue

訊息佇列，提供了FIFO的處理機制，具有快取訊息的能力。rabbitmq中，佇列訊息可以設定為持久化，臨時或者自動刪除。

  1. 設定為持久化的佇列，queue中的訊息會在server本地硬碟儲存一份，防止系統crash，資料丟失
  2. 設定為臨時佇列，queue中的資料在系統重啟之後就會丟失
  3. 設定為自動刪除的佇列，當不存在使用者連線到server，佇列中的資料會被自動刪除

Exchange

Exchange類似於資料通訊網路中的交換機，提供訊息路由策略。rabbitmq中，producer不是通過通道直接將訊息傳送給queue，而是先發送給Exchange。一個Exchange可以和多個Queue進行繫結，producer在傳遞訊息的時候，會傳遞一個ROUTING_KEY，Exchange會根據這個ROUTING_KEY按照特定的路由演算法，將訊息路由給指定的queue。和Queue一樣，Exchange也可設定為持久化，臨時或者自動刪除。

Exchange有4種類型：direct(預設)，fanout, topic, 和headers，不同型別的Exchange轉發訊息的策略有所區別：

  1. Direct直接交換器，工作方式類似於單播，Exchange會將訊息傳送完全匹配ROUTING_KEY的Queue
  2. fanout廣播是式交換器，不管訊息的ROUTING_KEY設定為什麼，Exchange都會將訊息轉發給所有繫結的Queue
  3. topic主題交換器，工作方式類似於組播，Exchange會將訊息轉發和ROUTING_KEY匹配模式相同的所有佇列，比如，ROUTING_KEY為user.stock的Message會轉發給繫結匹配模式為 * .stock,user.stock， * . * 和#.user.stock.#的佇列。（ * 表是匹配一個任意片語，#表示匹配0個或多個片語）
  4. headers訊息體的header匹配（ignore）

Binding

所謂繫結就是將一個特定的 Exchange 和一個特定的 Queue 繫結起來。Exchange 和Queue的繫結可以是多對多的關係

通訊過程

假設P1和C1註冊了相同的Broker，Exchange和Queue。P1傳送的訊息最終會被C1消費。基本的通訊流程大概如下所示：

  1. P1生產訊息，傳送給伺服器端的Exchange
  2. Exchange收到訊息，根據ROUTINKEY，將訊息轉發給匹配的Queue1
  3. Queue1收到訊息，將訊息傳送給訂閱者C1
  4. C1收到訊息，傳送ACK給佇列確認收到訊息
  5. Queue1收到ACK，刪除佇列中快取的此條訊息

Consumer收到訊息時需要顯式的向rabbit broker傳送basic.ack訊息或者consumer訂閱訊息時設定auto_ack引數為true。在通訊過程中，佇列對ACK的處理有以下幾種情況：

  1. 如果consumer接收了訊息，傳送ack,rabbitmq會刪除佇列中這個訊息，傳送另一條訊息給consumer。
  2. 如果cosumer接受了訊息, 但在傳送ack之前斷開連線，rabbitmq會認為這條訊息沒有被deliver,在consumer在次連線的時候，這條訊息會被redeliver。
  3. 如果consumer接受了訊息，但是程式中有bug,忘記了ack,rabbitmq不會重複傳送訊息。
  4. rabbitmq2.0.0和之後的版本支援consumer reject某條（類）訊息，可以通過設定requeue引數中的reject為true達到目地，那麼rabbitmq將會把訊息傳送給下一個註冊的consumer。

php 生產者示例

先用 composer 載入 mq 拓展檔案

{ 
　　"require": { 
　　　　"php-amqplib/php-amqplib": "2.7.*" //增加這行 
　　} 
}

 

<?php

require 'vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$conf = [
        'host' => '127.0.0.1',
        'port' => 5672,
        'user' => 'kd_dev',
        'pwd'  => 'kd_dev',
        'vhost' => '/',
];
$exchangeName = 'kd_sms_send_ex'; //交換機名
$queueName    = 'kd_sms_send_q'; //佇列名稱
$routingKey   = 'sms_send'; //路由關鍵字(也可以省略)

$conn = new AMQPStreamConnection( //建立生產者與mq之間的連線
    $conf['host'], $conf['port'], $conf['user'], $conf['pwd'], $conf['vhost']
);
$channel = $conn->channel(); //在已連線基礎上建立生產者與mq之間的通道


$channel->exchange_declare($exchangeName, 'direct', false, true, false); //宣告初始化交換機
$channel->queue_declare($queueName, false, true, false, false); //宣告初始化一條佇列
$channel->queue_bind($queueName, $exchangeName, $routingKey); //將佇列與某個交換機進行繫結，並使用路由關鍵字

$msgBody = json_encode(["name" => "iGoo", "age" => 22]);
$msg = new AMQPMessage($msgBody, ['content_type' => 'text/plain', 'delivery_mode' => 2]); //生成訊息
$r   = $channel->basic_publish($msg, $exchangeName, $routingKey); //推送訊息到某個交換機
$channel->close();
$conn->close();

 php 消費者程式碼示例

<?php

　　$bindingkey='sms_send'; 　　

　　//連線RabbitMQ
　　$conn_args = array( 'host'=>'127.0.0.1' , 'port'=> '5672', 'login'=>'kd_dev' , 'password'=> 'kd_dev','vhost' =>'/');
　　$conn = new AMQPConnection($conn_args);
　　$conn->connect(); 
　　
　　//設定queue名稱，使用exchange，繫結routingkey
　　$channel = new AMQPChannel($conn); // 宣告一個通道
　　$q = new AMQPQueue($channel);  // 宣告一個佇列
　　$q->setName('kd_sms_send_q'); // 路由名
　　$q->setFlags(AMQP_DURABLE);
 　 $q->declare();
 　 $q->bind('kd_sms_send_ex',$bindingkey); // 佇列繫結交換機 

　　//訊息獲取
　　$messages = $q->get(AMQP_AUTOACK) ;
　　if ($messages){
　　　　var_dump(json_decode($messages->getBody(), true ));
　　}

　　$conn->disconnect();

 

轉自：https://www.cnblogs.com/jun-ma/p/4840869.html

通訊過程

假設P1和C1註冊了相同的Broker，Exchange和Queue。P1傳送的訊息最終會被C1消費。基本的通訊流程大概如下所示：

  1. P1生產訊息，傳送給伺服器端的Exchange
  2. Exchange收到訊息，根據ROUTINKEY，將訊息轉發給匹配的Queue1
  3. Queue1收到訊息，將訊息傳送給訂閱者C1
  4. C1收到訊息，傳送ACK給佇列確認收到訊息
  5. Queue1收到ACK，刪除佇列中快取的此條訊息

Consumer收到訊息時需要顯式的向rabbit broker傳送basic.ack訊息或者consumer訂閱訊息時設定auto_ack引數為true。在通訊過程中，佇列對ACK的處理有以下幾種情況：

  1. 如果consumer接收了訊息，傳送ack,rabbitmq會刪除佇列中這個訊息，傳送另一條訊息給consumer。
  2. 如果cosumer接受了訊息, 但在傳送ack之前斷開連線，rabbitmq會認為這條訊息沒有被deliver,在consumer在次連線的時候，這條訊息會被redeliver。
  3. 如果consumer接受了訊息，但是程式中有bug,忘記了ack,rabbitmq不會重複傳送訊息。
  4. rabbitmq2.0.0和之後的版本支援consumer reject某條（類）訊息，可以通過設定requeue引數中的reject為true達到目地，那麼rabbitmq將會把訊息傳送給下一個註冊的consumer。

php 生產者示例

先用 composer 載入 mq 拓展檔案

{ 
　　"require": { 
　　　　"php-amqplib/php-amqplib": "2.7.*" //增加這行 
　　} 
}

 

<?php

require 'vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$conf = [
        'host' => '127.0.0.1',
        'port' => 5672,
        'user' => 'kd_dev',
        'pwd'  => 'kd_dev',
        'vhost' => '/',
];
$exchangeName = 'kd_sms_send_ex'; //交換機名
$queueName    = 'kd_sms_send_q'; //佇列名稱
$routingKey   = 'sms_send'; //路由關鍵字(也可以省略)

$conn = new AMQPStreamConnection( //建立生產者與mq之間的連線
    $conf['host'], $conf['port'], $conf['user'], $conf['pwd'], $conf['vhost']
);
$channel = $conn->channel(); //在已連線基礎上建立生產者與mq之間的通道


$channel->exchange_declare($exchangeName, 'direct', false, true, false); //宣告初始化交換機
$channel->queue_declare($queueName, false, true, false, false); //宣告初始化一條佇列
$channel->queue_bind($queueName, $exchangeName, $routingKey); //將佇列與某個交換機進行繫結，並使用路由關鍵字

$msgBody = json_encode(["name" => "iGoo", "age" => 22]);
$msg = new AMQPMessage($msgBody, ['content_type' => 'text/plain', 'delivery_mode' => 2]); //生成訊息
$r   = $channel->basic_publish($msg, $exchangeName, $routingKey); //推送訊息到某個交換機
$channel->close();
$conn->close();

 php 消費者程式碼示例

<?php

　　$bindingkey='sms_send'; 　　

　　//連線RabbitMQ
　　$conn_args = array( 'host'=>'127.0.0.1' , 'port'=> '5672', 'login'=>'kd_dev' , 'password'=> 'kd_dev','vhost' =>'/');
　　$conn = new AMQPConnection($conn_args);
　　$conn->connect(); 
　　
　　//設定queue名稱，使用exchange，繫結routingkey
　　$channel = new AMQPChannel($conn); // 宣告一個通道
　　$q = new AMQPQueue($channel);  // 宣告一個佇列
　　$q->setName('kd_sms_send_q'); // 路由名
　　$q->setFlags(AMQP_DURABLE);
 　 $q->declare();
 　 $q->bind('kd_sms_send_ex',$bindingkey); // 佇列繫結交換機 

　　//訊息獲取
　　$messages = $q->get(AMQP_AUTOACK) ;
　　if ($messages){
　　　　var_dump(json_decode($messages->getBody(), true ));
　　}

　　$conn->disconnect();

 

相關推薦

Active MQ(一)--訊息佇列基本概念介紹

1.概要 本系列將介紹Active MQ 的原理以及實現，在學習之前，先了解一些基本概念。 1.1.什麼是訊息佇列 訊息佇列（Message queue）是在訊息的傳輸過程中儲存訊息的容器。訊息佇列管理器在將訊息從它的源中繼到它的目標時充當中間人。佇列

kafka分散式訊息佇列 — 基本概念介紹

轉載源：https://www.cnblogs.com/lsx1993/p/4627465.html這個應該算是之前比較火熱的詞了，一直沒時間抽出來看看。一個新東西出來，肯定是為了解決某些問題，不然不會有它的市場。先簡單看下。官方介紹：分散式、分割槽、支援複製的日誌提交系統適

rabbitmq - (訊息佇列) 的基本原理介紹

介紹 MQ全稱為Message Queue, 是一種分散式應用程式的的通訊方法，它是消費-生產者模型的一個典型的代表，producer往訊息佇列中不斷寫入訊息，而另一端consumer則可以讀取或者訂閱佇列中的訊息。RabbitMQ是MQ產品的典型代表，是一款基於AMQP協議可複用的企業訊息系統 系統架構

RabbitMQ訊息佇列的基本原理

1.背景 RabbitMQ是一個由erlang開發的AMQP(Advanved Message Queue)的開源實現。 2.應用場景 2.1非同步處理 場景說明：使用者註冊後，需要發註冊郵件和註冊簡訊,傳統的做法有兩種1.序列的方式;2.並行的方

rabbitmq - (消息隊列) 的基本原理介紹

可復用 word conn 大致 hpa 服務 ges bind nor 介紹 MQ全稱為Message Queue, 是一種分布式應用程序的的通信方法，它是消費-生產者模型的一個典型的代表，producer往消息隊列中不斷寫入消息，而另一端consumer則可以讀取或者訂

RabbitMQ訊息佇列（一）: Detailed Introduction 詳細介紹

原文地址：http://blog.csdn.net/anzhsoft/article/details/19563091 1. 歷史     RabbitMQ是一個由erlang開發的AMQP（Advanced Message Queue ）的開源實現。AMQP 的

RabbitMQ訊息佇列介紹

RabbitMQ簡介 AMQP，即Advanced Message Queuing Protocol，高階訊息佇列協議，是應用層協議的一個開放標準，為面向訊息的中介軟體設計。訊息中介軟體主要用於元件之間的解耦，訊息的傳送者無需知道訊息使用者的存在，反之亦然。 AMQP的

RabbitMQ訊息佇列（一基本概念和常用命令）

常用命令 /etc/init.d/rabbitmq-server start|stop|restart|reload rabbitmqctl  add_vhost  vhostname  ##建立vhost rabbitmqctl   delete_vhost  v

rabbitmq訊息佇列原理

一、rabbitmq架構 RabbitMQ是一個流行的開源訊息佇列系統，是AMQP（高階訊息佇列協議）標準的實現，由以高效能、健壯、可伸縮性出名的Erlang語言開發，並繼承了這些優點。rabbitmq簡單架構如下： 上圖簡單展示了rabbitmq的架構

分散式訊息佇列kafka系列介紹 — 基本概念

這個應該算是之前比較火熱的詞了，一直沒時間抽出來看看。一個新東西出來，肯定是為了解決某些問題，不然不會有它的市場。先簡單看下。 官方介紹：分散式、分割槽、支援複製的日誌提交系統 適用場景：顧名思義，特別適合用於系統日誌的非同步記錄，對於資料穩定性、一致性、可靠性要求不高的

（二）RabbitMQ訊息佇列-RabbitMQ訊息佇列架構與基本概念

沒錯我還是沒有講怎麼安裝和寫一個HelloWord，不過快了，這一章我們先了解下RabbitMQ的基本概念。 RabbitMQ架構 說是架構其實更像是應用場景下的架構（自己畫的有點醜，勿嫌棄） 從圖中可以看出RabbitMQ主要由Exchange和Qu

消息服務百科全書——Kafka基本原理介紹

feature col 都是 指定 ESS 生成 接下來 另一個 追加 架構 1.1 總體架構 因為Kafka內在就是分布式的，一個Kafka集群通常包括多個代理。 為了均衡負載，將話題分成多個分區，每個代理存儲一或多個分區。多個生產者和消費者能夠同時生產和獲取消息。 一個

RabbitMQ 訊息佇列之 Exchange Types

寫在前面 RabbitMQ遵循AMQP 0-9-1協議 複製程式碼 AMQP 0-9-1協議簡介 訊息釋出到交換站，這通常被比作郵局或郵箱。然後交換器使用稱為繫結的規則將訊息副本分發到佇列。然後，AMQP代理將訊息傳遞給訂閱佇列的消費者，或者根據需要從佇列中獲取訊息。 釋出訊息時，釋出者可以指定各種

python 64式: 第3式、rabbitmq訊息佇列使用

topicProductor.py內容如下 #!/usr/bin/env python # -*- coding: utf-8 -*- import pika import sys ''' 問題： 實現基於rabbitmq的生產者和消費者，消費者可以支援繫結路由鍵為notification.

python之RabbitMQ訊息佇列

RabbitMQ：訊息佇列 PY裡的佇列有：執行緒QUEUE、程序QUEUE 程序queue可以用於父程序與子程序進行互動，或者同屬於一父程序下多個子程序進行互動，但如果是兩個獨立的程式，是不能用這個QUEUE進行通訊的。 兩個獨立的程式之間，要找一箇中間代理，比如可以用socket通訊

Rabbitmq---訊息佇列

一 . MQ:message queue 　　訊息佇列的作用: 　　1 通訊解耦 　　2 高峰限流 原理分析: 一開始,認證系統是強耦合的,A系統傳遞認證系統訊息接收計算結果的過程中 　　1 傳給認證系統 　　2 認證系統計算 　　3 返回計算結果 　　4 讀取A系統邏輯 只要當前計算

RabbitMQ 訊息佇列 - topic 模式分發訊息

推薦閱讀 https://blog.csdn.net/column/details/15500.html topic 模式 根據 Binding 指定的 RoutingKey, Exchange 對 key 進行模式匹配後投遞到相應的 Queue, 模式匹配時符號

RabbitMQ 訊息佇列 - fanout 模式分發訊息

推薦閱讀 https://blog.csdn.net/column/details/15500.html fanout 模式 將同一個 message 傳送到所有同該 Exchange 繫結的 queue, 只要 RoutingKey 是一樣, 這條訊息都會被投遞

RabbitMQ 訊息佇列 - direct 模式分發訊息

推薦閱讀 https://blog.csdn.net/column/details/15500.html direct 模式 根據 Binding 指定的 Routing Key, 將符合Key的訊息傳送到 Binding 的 Queue p_direc

sqlite基本原理介紹與操作

sqlite 基本介紹 引題 為什麼使用資料庫： 之前的學習中，我們已經可以儲存資料，變數，讀寫本地檔案(txt，csv). 1. 持久化，記憶體中的變數當程式重啟和電腦斷電丟失資料，而硬碟可以長時間，持久地儲存資料，資料儲存到硬碟上的過程叫持久化， 2資料庫更加專業強大