RabbitMQ入門詳解以及使用

時間  2020-05-06
標籤 rabbitmq 入門 詳解 以及 使用 欄目 RabbitMQ 简体版
原文   原文鏈接

目的：

　　RabbitMQ簡介

　　RabbitMQ安裝及使用

　　　　　Centos安裝java

　　　　　Docker安裝（今天選擇Docker安裝方法）
node

　　RabbitMQ快速入門

　　交換機

RabbitMQ簡介

 各大主流中間件對比

　　 ActiveMQ 是 Apache 出品，最流行的，能力強勁的開源消息總線，而且它一個徹底支持 J M S 規範的消息中間件。c++

其豐富的 API 、多種集羣構建模式使得他成爲業界老牌消息中間件，在中小型企業中應用普遍！docker

MQ 衡量指標：服務性能、數據存儲、集羣架構編程

 

Kafka：

 

RocketMQ是阿里開源的消息中間件，目前也已經孵化爲Apache頂級項目，它是純java開發，具備高吞吐量、高可用性、適合大規模分佈式系統 應用的特色。centos

RocketMQ思路起源於Kafka,它對消息的可靠傳輸及事務 性作了優化，目前在阿里集團被普遍應用於交易、充值、流計算、消息推 送、日誌流式處理、binglog分發等場景api

 

 

RabbitMQ是使用Erlang語言開發的開源消息隊列系統，基於AMQP協議 來實現。瀏覽器

AMQP的主要特徵是面向消息、隊列、路由（包括點對點和發佈 /訂閱）、可靠性、安全。AMQP協議更多用在企業系統內，安全

對數據_致 性、穩定性和可靠性要求很髙的場景，對性能和吞吐量的要求還在其次。服務器

 

結論：

activiMq老牌消息中間件，api全面，可是吞吐量不大 Kafaka吞吐量大，可是數據沒法保證不丟失，主要面向大數據 rokectMQ：吞吐量大，保證數據不丟失，而且支持分佈式事物，可是商業版須要收費 rabbitMQ：吞吐量大，數據不易丟失

 

初識RabbitMQ:

RabbitMQ是—個開源的消息代理和隊列服務器，用來經過普通協議 在徹底不一樣的應用之間共享數據，RabbitMQ是使用Erlang語言來編寫 的，而且RabbitMQ是基於AMQP協議的。

 

哪些大廠在用RabbitMQ,爲什幺？

滴滴、美團、頭條、去哪兒、藝龍......

開源、性能優秀，穩定性保障

提供可靠性消息投遞模式（confirm)、返回模式（return )

與SpringAMQP完美的整合、API豐富

集羣模式豐富，表達式配置，HA模式，鏡像隊列模型

保證數據不丟失的前提作到高可靠性、可用性

RabbitMQ高性能的緣由？

Erlang語言最初在於交換機領域的架構模式，這樣使得 RabbitMQ在Broker之間進行數據交互的性能是很是優秀的

Erlang的優勢：Erlang有着和原生Socket—樣的延遲

 

什麼是AMQP高級消息隊列協議？

　　AMQP定義：

是具備現代特徵的二進制協議； 是一個提供統一消息服務的應用層標準高級消息隊列協議； 是應用層協議的一個開放標準，爲面向消息的中間件設計；

 

AMQP核心概念（重點）

Server：又稱Broker，接受客戶端的鏈接，實現AMQP實體服務 Connection：鏈接：應用程序與Broker的網絡鏈接 Channel：網絡通道，幾乎全部的操做都在Channel中進行，Channel是進行消息讀寫的通道；客戶端可創建多個Channel，每一個Channel表明一個會話任務； Message：消息，服務器與應用程序之間傳遞的數據，由Properties和Body組成。Properties能夠對消息進行裝飾，好比消息的優先級、延遲等高級特性；Body則就是消息體內容； Virtual host：虛擬地址，用於進行邏輯隔離，最上層的消息路由；一個Virtual Host裏面能夠有若干個Exchange和Queue，
　　　　　　　　同一個Virtual Host裏面不能有相同名稱的Exchange或Queue； Exchange：交換機，交換消息，根據路由鍵轉發消息到綁定的隊列； Binding：Exchange和Queue之間的虛擬鏈接，binding中能夠包含routing key； Routing key：一個路由規則，虛擬機可用它來肯定如何路由一個特定消息 Queue：也稱爲Message Queue，消息隊列，保存消息並將它們轉發給消費者

 

 

 

 

RabbitMQ安裝及使用

 　　Centos安裝

 

 

 

 

 

注意：Erlang語言與RabbitMQ安裝版本必須匹配

　　RabbitMQ安裝與使用

官網地址：https://www.rabbitmq.com/

　　提早準備：安裝Linux必要依賴包

　　下載RabbitMQ必須安裝包

　　配置文件修改

　　服務的啓動：rabbitmq-server start &

　　服務的中止：rabbitmqctl stop_app

　　管理插件：rabbitmq-plugins enable rabbitmq_management

　　訪問地址：http://ip:15672/

詳細步驟：

準備： yum install \ build-essential openssl openssl-devel unixODBC unixODBC-devel \ make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz -y 下載： wget www.rabbitmq.com/releases/erlang/erlang-18.3-1.el7.centos.x86_64.rpm wget http://repo.iotti.biz/CentOS/7/x86_64/socat-1.7.3.2-5.el7.lux.x86_64.rpm
wget www.rabbitmq.com/releases/rabbitmq-server/v3.6.5/rabbitmq-server-3.6.5-1.noarch.rpm 安裝： rpm -ivh erlang-18.3-1.el7.centos.x86_64.rpm rpm -ivh socat-1.7.3.2-1.1.el7.x86_64.rpm  --nodeps --force rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm 配置文件： vi /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app 好比修改密碼、配置等等，例如：loopback_users 中的 <<"guest">>,只保留guest 服務啓動和中止： 啓動 rabbitmq-server start & 中止 rabbitmqctl app_stop 查看服務是否成功： yum install lsof lsof -i:5672 管理插件：rabbitmq-plugins enable rabbitmq_management 訪問地址：http://192.168.147.146:15672/

　　  Docker安裝

　　注意獲取鏡像的時候要獲取management版本的，不要獲取last版本的，management版本的才帶有管理界面

  • 獲取rabbitmq鏡像

docker pull rabbitmq:management

  • 查看rabbitmq鏡像

docker search rabbitmq:management

  • 運行rabbitmq鏡像

#方式一：默認guest用戶，密碼也是guest docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq:management #方式二：設置用戶名和密碼  docker run -d \ --name my-rabbitmq \ -p 5672:5672 -p 15672:15672 \ -v /data:/var/lib/rabbitmq \ --hostname my-rabbitmq-host \ -e RABBITMQ_DEFAULT_VHOST=my_vhost \ -e RABBITMQ_DEFAULT_USER=admin \ -e RABBITMQ_DEFAULT_PASS=admin \ --restart=always \ rabbitmq:management

 參數說明：

　 -d：後臺運行容器 -name：指定容器名 -p：指定服務運行的端口（5672：應用訪問端口；15672：控制檯Web端口號） -v：映射目錄或文件，啓動了一個數據卷容器，數據卷路徑爲：/var/lib/rabbitmq，再將此數據卷映射到住宿主機的/data目錄 --hostname：主機名（RabbitMQ的一個重要注意事項是它根據所謂的 「節點名稱」 存儲數據，默認爲主機名） -e：指定環境變量；（RABBITMQ_DEFAULT_VHOST：默認虛擬機名；RABBITMQ_DEFAULT_USER：默認的用戶名；RABBITMQ_DEFAULT_PASS：默認用戶名的密碼） --restart=always：當Docker重啓時，容器能自動啓動 rabbitmq:management：鏡像名  注1：RABBITMQ_DEFAULT_VHOST=my_vhost，my_vhost名字請記好，在以後的編程中要用到, 若是啓動時沒指定，默認值爲/

 

經常使用操做命令

 

  • 命令行與管控臺-基礎操做

 

rabbitmqctl stop_app：關閉應用 rabbitmqctl start_app：啓動應用 rabbitmqctl status：節點狀態 rabbitmqctl add_user username password：添加用戶 rabbitmqctl list_users：列出全部用戶 rabbitmqctl delete_user username：刪除用戶 rabbitmqctl clear_permissions -p vhostpath username：清除用戶權限 rabbitmqctl list_user_permissions username：列出用戶權限 rabbitmqctl change_password username newpassword：修改密碼 rabbitmqctl set_permissions -p vhostpath username 「.*」 「.*」 「.*」 rabbitmqctl add_vhost vhostpath：建立虛擬主機 rabbitmqctl list_vhosts：列出全部虛擬主機 rabbitmqctl list_permissions -p vhostpath：列出虛擬主機上全部權限 rabbitmqctl delete_vhost vhostpath：刪除虛擬主機 rabbitmqctl list_queues：查看全部隊列信息 rabbitmqctl -p vhostpath purge_queue blue：清除隊列裏的消息

  • 命令行與管控臺-高級操做

rabbitmqctl reset：移除全部數據，要在rabbitmqctl stop_app以後使用 rabbitmqctl join_cluster <clustermode> [--ram]：組成集羣命令 rabbitmqctl cluster_status：查看集羣狀態 rabbitmqctl change_cluster_node_type disc | ram：修改集羣節點的存儲形式 rabbitmqctl forget_cluster_node {--offline} 忘記節點 （摘除節點） rabbitmqctl rename_cluster_node oldnode1 newnode1 [oldnode2] [newnode2...] （修改節點名稱）

進入RabbitMQ管理平臺進行相關操做在瀏覽器中查看

輸入宿主機的ip+Rabbitmq的端口（15672）

 

我上面設置的是默認密碼帳號 guest，輸入登錄進去ok

 

Rabbitmq進去了，咱們下面來結合SpringCloud項目玩玩吧

 RabbitMQ快速入門

極速入門-消息生產與消費

      1. ConnectionFactory：獲取鏈接工廠
      2. Connection：一個連接
      3. Channel：數據通訊通道，課發送和接收消息
      4. Queue：具體的消息存儲隊列
      5. Producer & Consumer：生產和消費者

 

首先，在父類maven項目中建立一個SpringCloud項目，rabbitmq-api

 

在pom中添加下面一段依賴

<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>3.6.5</version> </dependency>

消費端代碼

package com.javaxh.rabbitmqapi.quickstar; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.QueueingConsumer; import java.io.IOException; public class Consumer { public static void main(String[] args) throws Exception { //1 建立一個ConnectionFactory, 並進行配置
        ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.239.130"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); //2 經過鏈接工廠建立鏈接
        Connection connection = connectionFactory.newConnection(); //3 經過connection建立一個Channel
        Channel channel = connection.createChannel(); //4 聲明（建立）一個隊列
        String queueName = "test001"; // 參數：隊列名稱、持久化與否、獨佔與否、無消息隊列是否自動刪除、消息參數 // queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        channel.queueDeclare(queueName, true, false, false, null); //5 建立消費者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel); //6 設置Channel // 參數：隊列名稱、自動簽收、消費者回調 // basicConsume(String queue, boolean autoAck, Consumer callback)
        channel.basicConsume(queueName, true, queueingConsumer); while(true){ //7 獲取消息（Delivery：傳送）
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery(); String msg = new String(delivery.getBody()); System.err.println("消費端: " + msg); //Envelope envelope = delivery.getEnvelope();
 } } }

生產端

package com.javaxh.rabbitmqapi.quickstar; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Procuder { public static void main(String[] args) throws Exception { //1 建立一個ConnectionFactory, 並進行配置
        ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.239.130"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); //2 經過鏈接工廠建立鏈接
        Connection connection = connectionFactory.newConnection(); //3 經過connection建立一個Channel
        Channel channel = connection.createChannel(); //4 經過Channel發送數據
        for(int i=0; i < 5; i++){ String msg = "Hello RabbitMQ!"; //1 exchange 2 routingKey
            channel.basicPublish("", "test001", null, msg.getBytes()); } //5 記得要關閉相關的鏈接
 channel.close(); connection.close(); } }

運行消費端，而後刷新RabbitMQ

交換機

 

 

交換機屬性：

Name：交換機名稱

Type：交換機類型 direct、topic、fanout、headers

Durability：是否須要持久化，true爲持久化

Auto Delete：當最後一個綁定到Exchange上的隊列刪除後，自動刪除該Exchange

Internal：當前Exchange是否用於RabbitMQ內部使用，默認爲False

Arguments：擴展參數，用於擴展AMQP協議，定製化使用

  • 直流交換機

直連交換機Direct Exchange（徹底匹配路由key）

全部發送到Direct Exchange的消息會被轉發到RouteKey中指定的Queue

 

注意：Direct模式可使用RabbitMQ自帶的Exchange：default Exchange，因此不須要將Exchange進行任何綁定（binding）操做，消息傳遞時，

RouteKey必須徹底匹配纔會被隊列接收，不然該消息會被拋棄；

 

 消費端代碼：

package com.javaxh.rabbitmqapi.direct; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.QueueingConsumer; public class Consumer4DirectExchange { public static void main(String[] args) throws Exception { ConnectionFactory connectionFactory = new ConnectionFactory() ; connectionFactory.setHost("192.168.239.131"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); connectionFactory.setAutomaticRecoveryEnabled(true); connectionFactory.setNetworkRecoveryInterval(3000); Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); //4 聲明
        String exchangeName = "test_direct_exchange"; String exchangeType = "direct"; String queueName = "test_direct_queue"; String routingKey = "test.direct"; //表示聲明瞭一個交換機
        channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null); //表示聲明瞭一個隊列
        channel.queueDeclare(queueName, false, false, false, null); //創建一個綁定關係:
 channel.queueBind(queueName, exchangeName, routingKey); //durable 是否持久化消息
        QueueingConsumer consumer = new QueueingConsumer(channel); //參數：隊列名稱、是否自動ACK、Consumer
        channel.basicConsume(queueName, true, consumer); //循環獲取消息
        while(true){ //獲取消息，若是沒有消息，這一步將會一直阻塞
            QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String msg = new String(delivery.getBody()); System.out.println("收到消息：" + msg); } } }

生產端代碼：

package com.javaxh.rabbitmqapi.direct; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Producer4DirectExchange { public static void main(String[] args) throws Exception { //1 建立ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.239.131"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); //2 建立Connection
        Connection connection = connectionFactory.newConnection(); //3 建立Channel
        Channel channel = connection.createChannel(); //4 聲明
        String exchangeName = "test_direct_exchange"; String routingKey = "test.direct"; // String routingKey = "test.direct111"; //收不到 //5 發送
 String msg = "Hello World RabbitMQ 4  Direct Exchange Message 111 ... "; channel.basicPublish(exchangeName, routingKey , null , msg.getBytes()); } }

啓動在刷新RabbitMQ康康

 

 

 

  •  主題交換機

主題交換機Topic Exchange（匹配路由規則的交換機）

 

全部發送到Topic Exchange的消息被轉發到全部關係RouteKey中指定Topic的Queue上

Exchange將RouteKey和某Topic進行模糊匹配，此時隊列須要綁定一個Topic

 

注意：可使用通配符進行模糊匹配

符號：「#」 匹配一個或者多個詞

符號：「*」 匹配很少很多一個詞

列如：

「log.#」 可以匹配到 「log.info.oa」

「log.*」 可以匹配到 「log.err」

 

 

 

消費端代碼

package com.javaxh.rabbitmqapi.topic; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.QueueingConsumer; public class Consumer4TopicExchange { public static void main(String[] args) throws Exception { ConnectionFactory connectionFactory = new ConnectionFactory() ; connectionFactory.setHost("192.168.239.131"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); connectionFactory.setAutomaticRecoveryEnabled(true); connectionFactory.setNetworkRecoveryInterval(3000); Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); //4 聲明
        String exchangeName = "test_topic_exchange"; String exchangeType = "topic"; String queueName = "test_topic_queue"; String routingKey = "user.#"; // String routingKey = "user.*"; // 1 聲明交換機
        channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null); // 2 聲明隊列
        channel.queueDeclare(queueName, false, false, false, null); // 3 創建交換機和隊列的綁定關係:
 channel.queueBind(queueName, exchangeName, routingKey); //durable 是否持久化消息
        QueueingConsumer consumer = new QueueingConsumer(channel); //參數：隊列名稱、是否自動ACK、Consumer
        channel.basicConsume(queueName, true, consumer); //循環獲取消息
        while(true){ //獲取消息，若是沒有消息，這一步將會一直阻塞
            QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String msg = new String(delivery.getBody()); System.out.println("收到消息：" + msg); } } }

生產端代碼

package com.javaxh.rabbitmqapi.topic; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Producer4TopicExchange { public static void main(String[] args) throws Exception { //1 建立ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.239.131"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); //2 建立Connection
        Connection connection = connectionFactory.newConnection(); //3 建立Channel
        Channel channel = connection.createChannel(); //4 聲明
        String exchangeName = "test_topic_exchange"; String routingKey1 = "user.save"; String routingKey2 = "user.update"; String routingKey3 = "user.delete.abc"; //5 發送
 String msg = "Hello World RabbitMQ 4 Topic Exchange Message ..."; channel.basicPublish(exchangeName, routingKey1 , null , msg.getBytes()); channel.basicPublish(exchangeName, routingKey2 , null , msg.getBytes()); channel.basicPublish(exchangeName, routingKey3 , null , msg.getBytes()); channel.close(); connection.close(); } }

啓動測試：

  • 輸出交換機

輸出交換機Fanout Exchange（不作路由）

不處理路由鍵，只須要簡單的將隊列綁定到交換機上；

發送到交換機的消息都會被轉發到與該交換機綁定的全部隊列上；

Fanout交換機轉發消息是最快的

 

 

消費端代碼

package com.javaxh.rabbitmqapi.fanout; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.QueueingConsumer; public class Consumer4FanoutExchange { public static void main(String[] args) throws Exception { ConnectionFactory connectionFactory = new ConnectionFactory() ; connectionFactory.setHost("192.168.239.131"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); connectionFactory.setAutomaticRecoveryEnabled(true); connectionFactory.setNetworkRecoveryInterval(3000); Connection connection = connectionFactory.newConnection(); Channel channel = connection.createChannel(); //4 聲明
        String exchangeName = "test_fanout_exchange"; String exchangeType = "fanout"; String queueName = "test_fanout_queue"; String routingKey = "";    //不設置路由鍵
        channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null); channel.queueDeclare(queueName, false, false, false, null); channel.queueBind(queueName, exchangeName, routingKey); //durable 是否持久化消息
        QueueingConsumer consumer = new QueueingConsumer(channel); //參數：隊列名稱、是否自動ACK、Consumer
        channel.basicConsume(queueName, true, consumer); //循環獲取消息
        while(true){ //獲取消息，若是沒有消息，這一步將會一直阻塞
            QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String msg = new String(delivery.getBody()); System.out.println("收到消息：" + msg); } } }

生產端代碼

package com.javaxh.rabbitmqapi.fanout; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Producer4FanoutExchange { public static void main(String[] args) throws Exception { //1 建立ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.239.130"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); //2 建立Connection
        Connection connection = connectionFactory.newConnection(); //3 建立Channel
        Channel channel = connection.createChannel(); //4 聲明
        String exchangeName = "test_fanout_exchange"; //5 發送
        for(int i = 0; i < 10; i ++) { String msg = "Hello World RabbitMQ 4 FANOUT Exchange Message ..."; channel.basicPublish(exchangeName, "", null , msg.getBytes()); } channel.close(); connection.close(); } }

啓動康康：

 

 

Binding-綁定

 

Exchange和Exchange、Queue之間的鏈接關係；

 

Binding中能夠包含RoutingKey或者參數

Queue-消息隊列

消息隊列，實際存儲消息數據

Durability：是否持久化

Durable：是，Transient：否

Auto delete：如選yes，表明當最後一個監聽被移除以後，該Queue會自動被刪除

Message-消息

服務器和應用程序之間傳遞的數據

本質上就是一段數據，由Properties和Payload（Body）組成

經常使用屬性：delivery model、headers（自定義屬性）

 

Message-其餘屬性

content_type、content_encoding、priority

correlation_id、reply_to、expiration、message_id

Timestamp、type、user_id、app_id、cluster_id

Virtual host-虛擬主機

 

虛擬地址，用於進行邏輯隔離，最上層的消息路由

 

一個Virtual Host裏面能夠有若干個Exchange和Queue

 

同一個Virtual Host裏面不能有相同名稱的Exchange或Queue

 

 

 

謝謝觀看！！

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程