RabbitMQ實戰教程

本文轉載自查看原文 2020-03-12 11:29 2624 java

1.什么是MQ
消息隊列（Message Queue，簡稱MQ），從字面意思上看，本質是個隊列，FIFO先入先出，只不過隊列中存放的內容是message而已。
其主要用途：不同進程Process/線程Thread之間通信。
為什么會產生消息隊列？有幾個原因：

不同進程（process）之間傳遞消息時，兩個進程之間耦合程度過高，改動一個進程，引發必須修改另一個進程，為了隔離這兩個進程，在兩進程間抽離出一層（一個模塊），所有兩進程之間傳遞的消息，都必須通過消息隊列來傳遞，單獨修改某一個進程，不會影響另一個；

不同進程（process）之間傳遞消息時，為了實現標准化，將消息的格式規范化了，並且，某一個進程接受的消息太多，一下子無法處理完，並且也有先后順序，必須對收到的消息進行排隊，因此誕生了事實上的消息隊列；

關於消息隊列的詳細介紹請參閱：
《Java帝國之消息隊列》
《一個故事告訴你什么是消息隊列》
《到底什么時候該使用MQ》

MQ框架非常之多，比較流行的有RabbitMq、ActiveMq、ZeroMq、kafka，以及阿里開源的RocketMQ。本文主要介紹RabbitMq。

2.RabbitMQ
2.1.RabbitMQ的簡介

開發語言：Erlang – 面向並發的編程語言。

2.1.1.AMQP
AMQP是消息隊列的一個協議。

2.2.官網

2.3.MQ的其他產品

2.4.學習5種隊列

2.5.安裝文檔

3.搭建RabbitMQ環境
3.1.下載
下載地址：http://www.rabbitmq.com/download.html

3.2.windows下安裝
3.2.1.安裝Erlang
下載：http://www.erlang.org/download/otp_win64_17.3.exe
安裝：

安裝完成。

3.2.2.安裝RabbitMQ

安裝完成。

開始菜單里出現如下選項：

啟動、停止、重新安裝等。

3.2.3.啟用管理工具
1、雙擊
2、進入C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.4.1\sbin輸入命令：
rabbitmq-plugins enable rabbitmq_management

這樣就啟動了管理工具，可以試一下命令：
停止：net stop RabbitMQ
啟動：net start RabbitMQ

3、在瀏覽器中輸入地址查看：http://127.0.0.1:15672/

4、使用默認賬號登錄：guest/ guest

3.3.Linux下安裝
3.3.1.安裝Erlang
3.3.2.添加yum支持
cd /usr/local/src/
mkdir rabbitmq
cd rabbitmq

wget http://packages.erlang-solutions ... ns-1.0-1.noarch.rpm
rpm -Uvh erlang-solutions-1.0-1.noarch.rpm

rpm --import http://packages.erlang-solutions.com/rpm/erlang_solutions.asc

使用yum安裝：
sudo yum install erlang

3.3.3.安裝RabbitMQ
上傳rabbitmq-server-3.4.1-1.noarch.rpm文件到/usr/local/src/rabbitmq/
安裝：
rpm -ivh rabbitmq-server-3.4.1-1.noarch.rpm

3.3.4.啟動、停止
service rabbitmq-server start
service rabbitmq-server stop
service rabbitmq-server restart
3.3.5.設置開機啟動
chkconfig rabbitmq-server on
3.3.6.設置配置文件
cd /etc/rabbitmq
cp /usr/share/doc/rabbitmq-server-3.4.1/rabbitmq.config.example /etc/rabbitmq/

mv rabbitmq.config.example rabbitmq.config
3.3.7.開啟用戶遠程訪問
vi /etc/rabbitmq/rabbitmq.config

注意要去掉后面的逗號。
3.3.8.開啟web界面管理工具
rabbitmq-plugins enable rabbitmq_management
service rabbitmq-server restart
3.3.9.防火牆開放15672端口
/sbin/iptables -I INPUT -p tcp --dport 15672 -j ACCEPT
/etc/rc.d/init.d/iptables save

3.4.安裝的注意事項
1、推薦使用默認的安裝路徑
2、系統用戶名必須是英文
Win10改名字非常麻煩，具體方法百度

3、計算機名必須是英文

4、系統的用戶必須是管理員

如果安裝失敗應該如何解決：
1、重裝系統 – 不推薦
2、將RabbitMQ安裝到linux虛擬機中
a)推薦
3、使用別人安裝好的RabbitMQ服務
a)只要給你開通一個賬戶即可。
b)使用公用的RabbitMQ服務，在192.168.50.22
c)推薦

常見錯誤：

3.5.安裝完成后操作
1、系統服務中有RabbitMQ服務，停止、啟動、重啟

2、打開命令行工具

如果找不到命令行工具,直接cd到相應目錄：

輸入命令rabbitmq-plugins enable rabbitmq_management啟用管理插件

查看管理頁面

通過默認賬戶 guest/guest 登錄
如果能夠登錄，說明安裝成功。

4.添加用戶
4.1.添加admin用戶

4.2.用戶角色
1、超級管理員(administrator)
可登陸管理控制台，可查看所有的信息，並且可以對用戶，策略(policy)進行操作。
2、監控者(monitoring)
可登陸管理控制台，同時可以查看rabbitmq節點的相關信息(進程數，內存使用情況，磁盤使用情況等)
3、策略制定者(policymaker)
可登陸管理控制台, 同時可以對policy進行管理。但無法查看節點的相關信息(上圖紅框標識的部分)。
4、普通管理者(management)
僅可登陸管理控制台，無法看到節點信息，也無法對策略進行管理。
5、其他
無法登陸管理控制台，通常就是普通的生產者和消費者。

4.3.創建Virtual Hosts

選中Admin用戶，設置權限：

看到權限已加：

4.4.管理界面中的功能

5.學習五種隊列

5.1.導入my-rabbitmq項目
項目下載地址：
https://download.csdn.net/download/zpcandzhj/10585077

5.2.簡單隊列
5.2.1.圖示

P：消息的生產者
C：消息的消費者
紅色：隊列

生產者將消息發送到隊列，消費者從隊列中獲取消息。
5.2.2.導入RabbitMQ的客戶端依賴

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

5.2.3.獲取MQ的連接

package com.zpc.rabbitmq.util;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;

public class ConnectionUtil {

public static Connection getConnection() throws Exception {
      //定義連接工廠
      ConnectionFactory factory = new ConnectionFactory();
      //設置服務地址
      factory.setHost("localhost");
      //端口
      factory.setPort(5672);
      //設置賬號信息，用戶名、密碼、vhost
      factory.setVirtualHost("testhost");
      factory.setUsername("admin");
      factory.setPassword("admin");
      // 通過工程獲取連接
      Connection connection = factory.newConnection();
      return connection;
}
}
5.2.4.生產者發送消息到隊列

package com.zpc.rabbitmq.simple;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Send {

private final static String QUEUE_NAME = "q_test_01";

public static void main(String[] argv) throws Exception {
      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      // 從連接中創建通道
      Channel channel = connection.createChannel();

      // 聲明（創建）隊列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);

      // 消息內容
      String message = "Hello World!";
      channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
      System.out.println(" [x] Sent '" + message + "'");
      //關閉通道和連接
      channel.close();
      connection.close();
}
}
5.2.5.管理工具中查看消息

點擊上面的隊列名稱，查詢具體的隊列中的信息：

5.2.6.消費者從隊列中獲取消息

package com.zpc.rabbitmq.simple;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

public class Recv {

private final static String QUEUE_NAME = "q_test_01";

public static void main(String[] argv) throws Exception {

      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      // 從連接中創建通道
      Channel channel = connection.createChannel();
      // 聲明隊列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);

      // 定義隊列的消費者
      QueueingConsumer consumer = new QueueingConsumer(channel);

      // 監聽隊列
      channel.basicConsume(QUEUE_NAME, true, consumer);

      // 獲取消息
      while (true) {
         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
         String message = new String(delivery.getBody());
         System.out.println(" [x] Received '" + message + "'");
      }
}
}

5.3.Work模式

5.3.1.圖示

一個生產者、2個消費者。

一個消息只能被一個消費者獲取。
5.3.2.消費者1

package com.zpc.rabbitmq.work;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv {

private final static String QUEUE_NAME = "test_queue_work";

public static void main(String[] argv) throws Exception {

      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      Channel channel = connection.createChannel();

      // 聲明隊列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);

      // 同一時刻服務器只會發一條消息給消費者
      //channel.basicQos(1);

      // 定義隊列的消費者
      QueueingConsumer consumer = new QueueingConsumer(channel);
      // 監聽隊列，false表示手動返回完成狀態，true表示自動
      channel.basicConsume(QUEUE_NAME, true, consumer);

      // 獲取消息
      while (true) {
         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
         String message = new String(delivery.getBody());
         System.out.println(" [y] Received '" + message + "'");
         //休眠
         Thread.sleep(10);
         // 返回確認狀態，注釋掉表示使用自動確認模式
         //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
}
}
5.3.3.消費者2

package com.zpc.rabbitmq.work;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv2 {

private final static String QUEUE_NAME = "test_queue_work";

public static void main(String[] argv) throws Exception {

      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      Channel channel = connection.createChannel();

      // 聲明隊列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);

      // 同一時刻服務器只會發一條消息給消費者
      //channel.basicQos(1);

      // 定義隊列的消費者
      QueueingConsumer consumer = new QueueingConsumer(channel);
      // 監聽隊列，false表示手動返回完成狀態，true表示自動
      channel.basicConsume(QUEUE_NAME, true, consumer);

      // 獲取消息
      while (true) {
         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
         String message = new String(delivery.getBody());
         System.out.println(" [x] Received '" + message + "'");
         // 休眠1秒
         Thread.sleep(1000);
         //下面這行注釋掉表示使用自動確認模式
         //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
}
}
5.3.4.生產者
向隊列中發送100條消息。

package com.zpc.rabbitmq.work;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Send {

private final static String QUEUE_NAME = "test_queue_work";

public static void main(String[] argv) throws Exception {
      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      Channel channel = connection.createChannel();

      // 聲明隊列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);

      for (int i = 0; i < 100; i++) {
         // 消息內容
         String message = "" + i;
         channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
         System.out.println(" [x] Sent '" + message + "'");

         Thread.sleep(i * 10);
      }

      channel.close();
      connection.close();
}
}

5.3.5.測試
測試結果：
1、消費者1和消費者2獲取到的消息內容是不同的，同一個消息只能被一個消費者獲取。
2、消費者1和消費者2獲取到的消息的數量是相同的，一個是消費奇數號消息，一個是偶數。

其實，這樣是不合理的，因為消費者1線程停頓的時間短。應該是消費者1要比消費者2獲取到的消息多才對。
RabbitMQ 默認將消息順序發送給下一個消費者，這樣，每個消費者會得到相同數量的消息。即輪詢（round-robin）分發消息。

怎樣才能做到按照每個消費者的能力分配消息呢？聯合使用 Qos 和 Acknowledge 就可以做到。
basicQos 方法設置了當前信道最大預獲取（prefetch）消息數量為1。消息從隊列異步推送給消費者，消費者的 ack 也是異步發送給隊列，從隊列的視角去看，總是會有一批消息已推送但尚未獲得 ack 確認，Qos 的 prefetchCount 參數就是用來限制這批未確認消息數量的。設為1時，隊列只有在收到消費者發回的上一條消息 ack 確認后，才會向該消費者發送下一條消息。prefetchCount 的默認值為0，即沒有限制，隊列會將所有消息盡快發給消費者。

2個概念

輪詢分發：使用任務隊列的優點之一就是可以輕易的並行工作。如果我們積壓了好多工作，我們可以通過增加工作者（消費者）來解決這一問題，使得系統的伸縮性更加容易。在默認情況下，RabbitMQ將逐個發送消息到在序列中的下一個消費者(而不考慮每個任務的時長等等，且是提前一次性分配，並非一個一個分配)。平均每個消費者獲得相同數量的消息。這種方式分發消息機制稱為Round-Robin（輪詢）。

公平分發：雖然上面的分配法方式也還行，但是有個問題就是：比如：現在有2個消費者，所有的奇數的消息都是繁忙的，而偶數則是輕松的。按照輪詢的方式，奇數的任務交給了第一個消費者，所以一直在忙個不停。偶數的任務交給另一個消費者，則立即完成任務，然后閑得不行。而RabbitMQ則是不了解這些的。這是因為當消息進入隊列，RabbitMQ就會分派消息。它不看消費者為應答的數目，只是盲目的將消息發給輪詢指定的消費者。

為了解決這個問題，我們使用basicQos( prefetchCount = 1)方法，來限制RabbitMQ只發不超過1條的消息給同一個消費者。當消息處理完畢后，有了反饋，才會進行第二次發送。
還有一點需要注意，使用公平分發，必須關閉自動應答，改為手動應答。

5.4.Work模式的“能者多勞”
打開上述代碼的注釋：

// 同一時刻服務器只會發一條消息給消費者
channel.basicQos(1);

//開啟這行表示使用手動確認模式
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

同時改為手動確認：

// 監聽隊列，false表示手動返回完成狀態，true表示自動
channel.basicConsume(QUEUE_NAME, false, consumer);
測試：
消費者1比消費者2獲取的消息更多。

5.5.消息的確認模式
消費者從隊列中獲取消息，服務端如何知道消息已經被消費呢？

模式1：自動確認
只要消息從隊列中獲取，無論消費者獲取到消息后是否成功消息，都認為是消息已經成功消費。
模式2：手動確認
消費者從隊列中獲取消息后，服務器會將該消息標記為不可用狀態，等待消費者的反饋，如果消費者一直沒有反饋，那么該消息將一直處於不可用狀態。

手動模式：

自動模式：

5.6.訂閱模式

5.6.1.圖示

解讀：
1、1個生產者，多個消費者
2、每一個消費者都有自己的一個隊列
3、生產者沒有將消息直接發送到隊列，而是發送到了交換機
4、每個隊列都要綁定到交換機
5、生產者發送的消息，經過交換機，到達隊列，實現，一個消息被多個消費者獲取的目的
注意：一個消費者隊列可以有多個消費者實例，只有其中一個消費者實例會消費

5.6.2.消息的生產者（看作是后台系統）
向交換機中發送消息。

package com.zpc.rabbitmq.subscribe;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

public class Send {

private final static String EXCHANGE_NAME = "test_exchange_fanout";

public static void main(String[] argv) throws Exception {
      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      Channel channel = connection.createChannel();

      // 聲明exchange
      channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

      // 消息內容
      String message = "Hello World!";
      channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
      System.out.println(" [x] Sent '" + message + "'");

      channel.close();
      connection.close();
}
}

注意：消息發送到沒有隊列綁定的交換機時，消息將丟失，因為，交換機沒有存儲消息的能力，消息只能存在在隊列中。
5.6.3.消費者1（看作是前台系統）

package com.zpc.rabbitmq.subscribe;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv {

private final static String QUEUE_NAME = "test_queue_work1";

private final static String EXCHANGE_NAME = "test_exchange_fanout";

public static void main(String[] argv) throws Exception {

      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      Channel channel = connection.createChannel();

      // 聲明隊列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);

      // 綁定隊列到交換機
      channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

      // 同一時刻服務器只會發一條消息給消費者
      channel.basicQos(1);

      // 定義隊列的消費者
      QueueingConsumer consumer = new QueueingConsumer(channel);
      // 監聽隊列，手動返回完成
      channel.basicConsume(QUEUE_NAME, false, consumer);

      // 獲取消息
      while (true) {
         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
         String message = new String(delivery.getBody());
         System.out.println(" [Recv] Received '" + message + "'");
         Thread.sleep(10);

         channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
}
}

5.6.4.消費者2（看作是搜索系統）

package com.zpc.rabbitmq.subscribe;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv2 {

private final static String QUEUE_NAME = "test_queue_work2";

private final static String EXCHANGE_NAME = "test_exchange_fanout";

public static void main(String[] argv) throws Exception {

      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      Channel channel = connection.createChannel();

      // 聲明隊列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);

      // 綁定隊列到交換機
      channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

      // 同一時刻服務器只會發一條消息給消費者
      channel.basicQos(1);

      // 定義隊列的消費者
      QueueingConsumer consumer = new QueueingConsumer(channel);
      // 監聽隊列，手動返回完成
      channel.basicConsume(QUEUE_NAME, false, consumer);

      // 獲取消息
      while (true) {
         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
         String message = new String(delivery.getBody());
         System.out.println(" [Recv2] Received '" + message + "'");
         Thread.sleep(10);

         channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
}
}

5.6.5.測試
測試結果：
同一個消息被多個消費者獲取。一個消費者隊列可以有多個消費者實例，只有其中一個消費者實例會消費到消息。

在管理工具中查看隊列和交換機的綁定關系：

5.7.路由模式

5.7.1.圖示

5.7.2.生產者

5.7.3.消費者1(假設是前台系統)

5.7.4.消費2（假設是搜索系統）

5.8.主題模式（通配符模式）

5.8.1.圖示

同一個消息被多個消費者獲取。一個消費者隊列可以有多個消費者實例，只有其中一個消費者實例會消費到消息。

5.8.2.生產者

package com.zpc.rabbitmq.topic;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Send {

private final static String EXCHANGE_NAME = "test_exchange_topic";

public static void main(String[] argv) throws Exception {
      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      Channel channel = connection.createChannel();

      // 聲明exchange
      channel.exchangeDeclare(EXCHANGE_NAME, "topic");

      // 消息內容
      String message = "Hello World!!";
      channel.basicPublish(EXCHANGE_NAME, "routekey.1", null, message.getBytes());
      System.out.println(" [x] Sent '" + message + "'");

      channel.close();
      connection.close();
}
}
5.8.3.消費者1（前台系統）

package com.zpc.rabbitmq.topic;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

import com.zpc.rabbitmq.util.ConnectionUtil;

public class Recv {

private final static String QUEUE_NAME = "test_queue_topic_work_1";

private final static String EXCHANGE_NAME = "test_exchange_topic";

public static void main(String[] argv) throws Exception {

      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      Channel channel = connection.createChannel();

      // 聲明隊列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);

      // 綁定隊列到交換機
      channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routekey.*");

      // 同一時刻服務器只會發一條消息給消費者
      channel.basicQos(1);

      // 定義隊列的消費者
      QueueingConsumer consumer = new QueueingConsumer(channel);
      // 監聽隊列，手動返回完成
      channel.basicConsume(QUEUE_NAME, false, consumer);

      // 獲取消息
      while (true) {
         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
         String message = new String(delivery.getBody());
         System.out.println(" [Recv_x] Received '" + message + "'");
         Thread.sleep(10);

         channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
}
}
5.8.4.消費者2（搜索系統）

package com.zpc.rabbitmq.topic;

import com.zpc.rabbitmq.util.ConnectionUtil;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;

public class Recv2 {

private final static String QUEUE_NAME = "test_queue_topic_work_2";

private final static String EXCHANGE_NAME = "test_exchange_topic";

public static void main(String[] argv) throws Exception {

      // 獲取到連接以及mq通道
      Connection connection = ConnectionUtil.getConnection();
      Channel channel = connection.createChannel();

      // 聲明隊列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);

      // 綁定隊列到交換機
      channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*");

      // 同一時刻服務器只會發一條消息給消費者
      channel.basicQos(1);

      // 定義隊列的消費者
      QueueingConsumer consumer = new QueueingConsumer(channel);
      // 監聽隊列，手動返回完成
      channel.basicConsume(QUEUE_NAME, false, consumer);

      // 獲取消息
      while (true) {
         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
         String message = new String(delivery.getBody());
         System.out.println(" [Recv2_x] Received '" + message + "'");
         Thread.sleep(10);

         channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      }
}
}

6.Spring-Rabbit
6.1.Spring項目
http://spring.io/projects

6.2.簡介

6.3.使用
6.3.1.消費者

package com.zpc.rabbitmq.spring;

/**
* 消費者
*
* @author Evan
*/
public class Foo {

//具體執行業務的方法
public void listen(String foo) {
      System.out.println("\n消費者： " + foo + "\n");
}
}

6.3.2.生產者

package com.zpc.rabbitmq.spring;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class SpringMain {
public static void main(final String... args) throws Exception {
      AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
            "classpath:spring/rabbitmq-context.xml");
      //RabbitMQ模板
      RabbitTemplate template = ctx.getBean(RabbitTemplate.class);
      //發送消息
      template.convertAndSend("Hello, 鳥鵬!");
      Thread.sleep(1000);// 休眠1秒
      ctx.destroy(); //容器銷毀
}
}
6.3.3.配置文件
1、定義連接工廠


<rabbit:connection-factory id="connectionFactory"
host="127.0.0.1" port="5672" username="admin" password="admin"
virtual-host="testhost" />

2、定義模板（可以指定交換機或隊列）

<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="fanoutExchange" />
1
3、定義隊列、交換機、以及完成隊列和交換機的綁定


<rabbit:queue name="zpcQueue" auto-declare="true"/>


<rabbit:fanout-exchange name="fanoutExchange" auto-declare="true">
<rabbit:bindings>
   <rabbit:binding queue="zpcQueue"/>
</rabbit:bindings>
</rabbit:fanout-exchange>

4、定義監聽

<rabbit:listener-container connection-factory="connectionFactory">
<rabbit:listener ref="foo" method="listen" queue-names="zpcQueue" />
</rabbit:listener-container>

<bean id="foo" class="com.zpc.rabbitmq.spring.Foo" />

5、定義管理，用於管理隊列、交換機等：


<rabbit:admin connection-factory="connectionFactory" />

完整配置文件rabbitmq-context.xml

<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit
http://www.springframework.org/s ... ring-rabbit-1.4.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">


<rabbit:connection-factory id="connectionFactory"
   host="127.0.0.1" port="5672" username="admin" password="admin"
   virtual-host="testhost" />


<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="fanoutExchange" />



<rabbit:admin connection-factory="connectionFactory" />


<rabbit:queue name="zpcQueue" auto-declare="true"/>


<rabbit:fanout-exchange name="fanoutExchange" auto-declare="true">
   <rabbit:bindings>
      <rabbit:binding queue="zpcQueue"/>
   </rabbit:bindings>
</rabbit:fanout-exchange>




<rabbit:listener-container connection-factory="connectionFactory">
   <rabbit:listener ref="foo" method="listen" queue-names="zpcQueue" />
</rabbit:listener-container>

<bean id="foo" class="com.zpc.rabbitmq.spring.Foo" />

</beans>

6.4.持久化交換機和隊列

持久化：將交換機或隊列的數據保存到磁盤，服務器宕機或重啟之后依然存在。
非持久化：將交換機或隊列的數據保存到內存，服務器宕機或重啟之后將不存在。

非持久化的性能高於持久化。

如何選擇持久化？非持久化？ – 看需求。

7.Spring集成RabbitMQ一個完整案例
創建三個系統A,B,C
A作為生產者，B、C作為消費者(B,C作為web項目啟動)
項目下載地址：https://download.csdn.net/download/zpcandzhj/10585077

7.1.在A系統中發送消息到交換機
7.1.1.導入依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>

<groupId>com.zpc</groupId>
<artifactId>myrabbitA</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>myrabbit</name>

<dependencies>
   <dependency>
      <groupId>org.springframework.amqp</groupId>
      <artifactId>spring-rabbit</artifactId>
      <version>1.4.0.RELEASE</version>
   </dependency>

   <dependency>
      <groupId>com.alibaba</groupId>
      <artifactId>fastjson</artifactId>
      <version>1.2.47</version>
   </dependency>
</dependencies>
</project>

7.1.2.隊列和交換機的綁定關系
實現：
1、在配置文件中將隊列和交換機完成綁定
2、可以在管理界面中完成綁定
a)綁定關系如果發生變化，需要修改配置文件，並且服務需要重啟
b)管理更加靈活
c)更容易對綁定關系的權限管理，流程管理
本例選擇第2種方式
7.1.3.配置
rabbitmq-context.xml

<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit
http://www.springframework.org/s ... ring-rabbit-1.4.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">


<rabbit:connection-factory id="connectionFactory"
   host="127.0.0.1" port="5672" username="admin" password="admin"
   virtual-host="testhost" />


<rabbit:admin connection-factory="connectionFactory" />



<rabbit:direct-exchange name="directExchange" auto-declare="true" ></rabbit:direct-exchange>




<rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="directExchange" />

</beans>

7.1.4.消息內容
方案：
1、消息內容使用對象做json序列化發送
a)數據大
b)有些數據其他人是可能用不到的
2、發送特定的業務字段，如id、操作類型

7.1.5.實現
生產者MsgSender.java：

package com.zpc.myrabbit;

import com.alibaba.fastjson.JSON;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
* 消息生產者
*/
public class MsgSender {
public static void main(String[] args) throws Exception {
      AbstractApplicationContext ctx = new ClassPathXmlApplicationContext(
            "classpath:spring/rabbitmq-context.xml");
      //RabbitMQ模板
      RabbitTemplate template = ctx.getBean(RabbitTemplate.class);

      String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小時制
      //發送消息
      Map<String, Object> msg = new HashMap<String, Object>();
      msg.put("type", "1");
      msg.put("date", date);
      template.convertAndSend("type2", JSON.toJSONString(msg));
      Thread.sleep(1000);// 休眠1秒
      ctx.destroy(); //容器銷毀
}
}
7.2.在B系統接收消息
7.2.1.導入依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>

<groupId>com.zpc</groupId>
<artifactId>myrabbitB</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>war</packaging>

<name>myrabbit</name>
<properties>
      <spring.version>4.1.3.RELEASE</spring.version>
      <fastjson.version>1.2.46</fastjson.version>
</properties>

<dependencies>
      <dependency>
         <groupId>com.rabbitmq</groupId>
         <artifactId>amqp-client</artifactId>
         <version>3.4.1</version>
      </dependency>
      <dependency>
         <groupId>org.springframework.amqp</groupId>
         <artifactId>spring-rabbit</artifactId>
         <version>1.4.0.RELEASE</version>
      </dependency>
      <dependency>
         <groupId>org.springframework</groupId>
         <artifactId>spring-webmvc</artifactId>
         <version>${spring.version}</version>
      </dependency>
      <dependency>
         <groupId>com.alibaba</groupId>
         <artifactId>fastjson</artifactId>
         <version>1.2.47</version>
      </dependency>
</dependencies>

<build>
      <finalName>${project.artifactId}</finalName>
      <plugins>
         
         <plugin>
            <groupId>org.apache.tomcat.maven</groupId>
            <artifactId>tomcat7-maven-plugin</artifactId>
            <configuration>
                  <path>/testRabbit</path>
                  <uriEncoding>UTF-8</uriEncoding>
                  <port>8081</port>
            </configuration>
         </plugin>
      </plugins>
</build>
</project>
7.2.2.配置

<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit
http://www.springframework.org/s ... ring-rabbit-1.4.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">


<rabbit:connection-factory id="connectionFactory"
   host="127.0.0.1" port="5672" username="admin" password="admin"
   virtual-host="testhost" />


<rabbit:admin connection-factory="connectionFactory" />


<rabbit:queue name="q_topic_testB" auto-declare="true"/>


<rabbit:listener-container connection-factory="connectionFactory">
   <rabbit:listener ref="myMQlistener" method="listen" queue-names="q_topic_testB" />
</rabbit:listener-container>

<bean id="myMQlistener" class="com.zpc.myrabbit.listener.Listener" />
</beans>
7.2.3.具體處理邏輯

public class Listener {
//具體執行業務的方法
public void listen(String msg) {
      System.out.println("\n消費者B開始處理消息： " + msg + "\n");
}
}
7.2.4.在界面管理工具中完成綁定關系
選中定義好的交換機(exchange)

1）direct

2）fanout

3）topic

7.3.在C系統中接收消息
（和B系統配置差不多，無非是Q名和Q對應的處理邏輯變了）

7.3.1.配置

<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/rabbit
http://www.springframework.org/s ... ring-rabbit-1.4.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.1.xsd">


<rabbit:connection-factory id="connectionFactory"
   host="127.0.0.1" port="5672" username="admin" password="admin"
   virtual-host="testhost" />


<rabbit:admin connection-factory="connectionFactory" />


<rabbit:queue name="q_topic_testC" auto-declare="true"/>


<rabbit:listener-container connection-factory="connectionFactory">
   <rabbit:listener ref="myMQlistener" method="listen" queue-names="q_topic_testC" />
</rabbit:listener-container>

<bean id="myMQlistener" class="com.zpc.myrabbit.listener.Listener" />
</beans>
7.3.2.處理業務邏輯

public class Listener {

//具體執行業務的方法
public void listen(String msg) {
      System.out.println("\n消費者C開始處理消息： " + msg + "\n");
}
}

7.3.3.在管理工具中綁定隊列和交換機
見7.2.4

7.3.4.測試
分別啟動B,C兩個web應用，然后運行A的MsgSender主方法發送消息，分別測試fanout、direct、topic三種類型

8.Springboot集成RabbitMQ
springboot集成RabbitMQ非常簡單，如果只是簡單的使用配置非常少，springboot提供了spring-boot-starter-amqp對消息各種支持。
代碼下載地址：https://download.csdn.net/download/zpcandzhj/10585077
8.1.簡單隊列
1、配置pom文件，主要是添加spring-boot-starter-amqp的支持

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2、配置application.properties文件
配置rabbitmq的安裝地址、端口以及賬戶信息

spring.application.name=spirng-boot-rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
3、配置隊列

package com.zpc.rabbitmq;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {
@Bean
public Queue queue() {
      return new Queue("q_hello");
}
}
4、發送者

package com.zpc.rabbitmq;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

@Component
public class HelloSender {
@Autowired
private AmqpTemplate rabbitTemplate;

public void send() {
      String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小時制
      String context = "hello " + date;
      System.out.println("Sender : " + context);
      //簡單對列的情況下routingKey即為Q名
      this.rabbitTemplate.convertAndSend("q_hello", context);
}
}
5、接收者

package com.zpc.rabbitmq;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "q_hello")
public class HelloReceiver {

@RabbitHandler
public void process(String hello) {
      System.out.println("Receiver  : " + hello);
}
}
6、測試

package com.zpc.rabbitmq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqHelloTest {

@Autowired
private HelloSender helloSender;

@Test
public void hello() throws Exception {
      helloSender.send();
}
}
8.2.多對多使用（Work模式）
注冊兩個Receiver:

package com.zpc.rabbitmq;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "q_hello")
public class HelloReceiver2 {

@RabbitHandler
public void process(String hello) {
      System.out.println("Receiver2  : " + hello);
}

}
@Test
public void oneToMany() throws Exception {
for (int i=0;i<100;i++){
      helloSender.send(i);
      Thread.sleep(300);
}
}
public void send(int i) {
String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());//24小時制
String context = "hello " + i + " " + date;
System.out.println("Sender : " + context);
//簡單對列的情況下routingKey即為Q名
this.rabbitTemplate.convertAndSend("q_hello", context);
}
8.3.Topic Exchange（主題模式）
topic 是RabbitMQ中最靈活的一種方式，可以根據routing_key自由的綁定不同的隊列
首先對topic規則配置，這里使用兩個隊列(消費者)來演示。
1)配置隊列，綁定交換機

package com.zpc.rabbitmq.topic;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TopicRabbitConfig {

final static String message = "q_topic_message";
final static String messages = "q_topic_messages";

@Bean
public Queue queueMessage() {
      return new Queue(TopicRabbitConfig.message);
}

@Bean
public Queue queueMessages() {
      return new Queue(TopicRabbitConfig.messages);
}

/**
   * 聲明一個Topic類型的交換機
   * @return
   */
@Bean
TopicExchange exchange() {
      return new TopicExchange("mybootexchange");
}

/**
   * 綁定Q到交換機,並且指定routingKey
   * @param queueMessage
   * @param exchange
   * @return
   */
@Bean
Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
      return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
}

@Bean
Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
      return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
}
}
2)創建2個消費者
q_topic_message 和q_topic_messages

package com.zpc.rabbitmq.topic;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "q_topic_message")
public class Receiver1 {

@RabbitHandler
public void process(String hello) {
      System.out.println("Receiver1  : " + hello);
}
}
package com.zpc.rabbitmq.topic;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "q_topic_messages")
public class Receiver2 {

@RabbitHandler
public void process(String hello) {
      System.out.println("Receiver2 : " + hello);
}
}
3)消息發送者（生產者）

package com.zpc.rabbitmq.topic;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MsgSender {

@Autowired
private AmqpTemplate rabbitTemplate;

public void send1() {
      String context = "hi, i am message 1";
      System.out.println("Sender : " + context);
      this.rabbitTemplate.convertAndSend("mybootexchange", "topic.message", context);
}

public void send2() {
      String context = "hi, i am messages 2";
      System.out.println("Sender : " + context);
      this.rabbitTemplate.convertAndSend("mybootexchange", "topic.messages", context);
}
}
send1方法會匹配到topic.#和topic.message，兩個Receiver都可以收到消息，發送send2只有topic.#可以匹配所有只有Receiver2監聽到消息。
4)測試

package com.zpc.rabbitmq.topic;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitTopicTest {

@Autowired
private MsgSender msgSender;

@Test
public void send1() throws Exception {
      msgSender.send1();
}

@Test
public void send2() throws Exception {
      msgSender.send2();
}
}
8.4.Fanout Exchange（訂閱模式）
Fanout 就是我們熟悉的廣播模式或者訂閱模式，給Fanout交換機發送消息，綁定了這個交換機的所有隊列都收到這個消息。
1)配置隊列，綁定交換機
package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutRabbitConfig {

@Bean
public Queue aMessage() {
      return new Queue("q_fanout_A");
}

@Bean
public Queue bMessage() {
      return new Queue("q_fanout_B");
}

@Bean
public Queue cMessage() {
      return new Queue("q_fanout_C");
}

@Bean
FanoutExchange fanoutExchange() {
      return new FanoutExchange("mybootfanoutExchange");
}

@Bean
Binding bindingExchangeA(Queue aMessage, FanoutExchange fanoutExchange) {
      return BindingBuilder.bind(aMessage).to(fanoutExchange);
}

@Bean
Binding bindingExchangeB(Queue bMessage, FanoutExchange fanoutExchange) {
      return BindingBuilder.bind(bMessage).to(fanoutExchange);
}

@Bean
Binding bindingExchangeC(Queue cMessage, FanoutExchange fanoutExchange) {
      return BindingBuilder.bind(cMessage).to(fanoutExchange);
}
}
2）創建3個消費者

package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "q_fanout_A")
public class ReceiverA {

@RabbitHandler
public void process(String hello) {
      System.out.println("AReceiver  : " + hello + "/n");
}
}
package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "q_fanout_B")
public class ReceiverB {

@RabbitHandler
public void process(String hello) {
      System.out.println("BReceiver  : " + hello + "/n");
}
}
package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "q_fanout_C")
public class ReceiverC {

@RabbitHandler
public void process(String hello) {
      System.out.println("CReceiver  : " + hello + "/n");
}
}
3）生產者

package com.zpc.rabbitmq.fanout;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MsgSenderFanout {

@Autowired
private AmqpTemplate rabbitTemplate;

public void send() {
      String context = "hi, fanout msg ";
      System.out.println("Sender : " + context);
      this.rabbitTemplate.convertAndSend("mybootfanoutExchange","", context);
}
}

4）測試

package com.zpc.rabbitmq.fanout;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitFanoutTest {

@Autowired
private MsgSenderFanout msgSender;

@Test
public void send1() throws Exception {
      msgSender.send();
}
}
結果如下，三個消費者都收到消息：
AReceiver : hi, fanout msg
CReceiver : hi, fanout msg
BReceiver : hi, fanout msg

9.總結
使用MQ實現商品數據的同步優勢：
1、降低系統間耦合度
2、便於管理數據的同步（數據一致性）

更多java學習資料可關注:itheimaGZ獲取

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 RabbitMQ實戰教程 RabbitMQ實戰 SpringBoot進階教程(五十)集成RabbitMQ---MQ實戰演練 rabbitmq 命令&& rabbitmq教程（一）【RabbitMQ 實戰指南】一 RabbitMQ入門【RabbitMQ 實戰指南】一 RabbitMQ 開發 RabbitMQ封裝實戰 rabbitMq實戰使用 RabbitMQ實戰應用技巧 RabbitMQ實戰指南PDF