RabbitMQ提供了6種消息模型，但是第6種其實是RPC，並不是MQ，因此不予學習。那么也就剩下5種。但是其實3、4、5這三種都屬於訂閱模型，只不過進行路由的方式不同。
0. 配置項目
pom.xml
<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.2.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>demo-rabbitmq</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>demo-rabbitmq</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.3.2</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
我們抽取一個建立RabbitMQ連接的工具類，方便其他程序獲取連接：
package com.example.demorabbitmq.util;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* @author john
* @date 2019/12/11 - 15:33
*/
public class ConnectionUtil {
/**
* 建立與RabbitMQ的連接
*
* @return
* @throws Exception
*/
public static Connection getConnection() throws Exception {
//定義連接工廠
ConnectionFactory factory = new ConnectionFactory();
//設置服務地址
factory.setHost("192.168.0.22");
//端口
factory.setPort(5672);
//設置賬號信息，用戶名、密碼、vhost
factory.setVirtualHost("/leyou");
factory.setUsername("leyou");
factory.setPassword("leyou");
// 通過工程獲取連接
Connection connection = factory.newConnection();
return connection;
}
}
1. 基本消息模型
RabbitMQ是一個消息代理：它接受和轉發消息。 你可以把它想象成一個郵局：當你把郵件放在郵箱里時，你可以確定郵差先生最終會把郵件發送給你的收件人。 在這個比喻中，RabbitMQ是郵政信箱，郵局和郵遞員。
RabbitMQ與郵局的主要區別是它不處理紙張，而是接受，存儲和轉發數據消息的二進制數據塊。
P（producer/ publisher）：生產者，一個發送消息的用戶應用程序。
C（consumer）：消費者，消費和接收有類似的意思，消費者是一個主要用來等待接收消息的用戶應用程序
隊列（紅色區域）：rabbitmq內部類似於郵箱的一個概念。雖然消息流經rabbitmq和你的應用程序，但是它們只能存儲在隊列中。隊列只受主機的內存和磁盤限制，實質上是一個大的消息緩沖區。許多生產者可以發送消息到一個隊列，許多消費者可以嘗試從一個隊列接收數據。
總之：
生產者將消息發送到隊列，消費者從隊列中獲取消息，隊列是存儲消息的緩沖區。
我們將用Java編寫兩個程序;發送單個消息的生產者，以及接收消息並將其打印出來的消費者。我們將詳細介紹Java API中的一些細節，這是一個消息傳遞的“Hello World”。
我們將調用我們的消息發布者（發送者）Send和我們的消息消費者（接收者）Recv。發布者將連接到RabbitMQ，發送一條消息，然后退出。
1.1 生產者發送消息
package com.example.demorabbitmq.simple;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
* 生產者
*/
public class Send {
private final static String QUEUE_NAME = "simple_queue_demo";
public static void main(String[] argv) throws Exception {
// 獲取到連接
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();
}
}
進入隊列頁面，可以看到新建了一個隊列：simple_queue_demo
點擊隊列名稱，進入詳情頁，可以查看消息：
在控制台查看消息並不會將消息消費，所以消息還在。
1.2 消費者獲取消息（自動ACK）
package com.example.demorabbitmq.simple;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* 消費者
*/
public class Recv {
private final static String QUEUE_NAME = "simple_queue_demo";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 創建通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [x] received : " + msg + "!");
}
};
// 監聽隊列，第二個參數：是否自動進行消息確認。
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
這個時候，隊列中的消息就沒了：
我們發現，消費者已經獲取了消息，但是程序沒有停止，一直在監聽隊列中是否有新的消息。一旦有新的消息進入隊列，就會立即打印.
1.3 消息確認機制（ACK）
通過剛才的案例可以看出，消息一旦被消費者接收，隊列中的消息就會被刪除。
那么問題來了：RabbitMQ怎么知道消息被接收了呢？
如果消費者領取消息后，還沒執行操作就掛掉了呢？或者拋出了異常？消息消費失敗，但是RabbitMQ無從得知，這樣消息就丟失了！
因此，RabbitMQ有一個ACK機制。當消費者獲取消息后，會向RabbitMQ發送回執ACK，告知消息已經被接收。不過這種回執ACK分兩種情況：
- 自動ACK：消息一旦被接收，消費者自動發送ACK
- 手動ACK：消息接收后，不會發送ACK，需要手動調用
大家覺得哪種更好呢？
這需要看消息的重要性：
- 如果消息不太重要，丟失也沒有影響，那么自動ACK會比較方便
- 如果消息非常重要，不容丟失。那么最好在消費完成后手動ACK，否則接收消息后就自動ACK，RabbitMQ就會把消息從隊列中刪除。如果此時消費者宕機，那么消息就丟失了。
我們之前的測試都是自動ACK的，如果要手動ACK，需要改動我們的代碼：
1.4 消費者獲取消息（手動ACK）
package com.example.demorabbitmq.simple;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* 消費者,手動進行ACK
*/
public class Recv2 {
private final static String QUEUE_NAME = "simple_queue_demo";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 創建通道
final Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [x] received : " + msg + "!");
// 手動進行ACK
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
// 監聽隊列，第二個參數false，手動進行ACK
channel.basicConsume(QUEUE_NAME, false, consumer);
}
}
注意到最后一行代碼：
channel.basicConsume(QUEUE_NAME, false, consumer);
如果第二個參數為true，則會自動進行ACK；如果為false，則需要手動ACK。則需要手動ACK。方法的聲明：
1.5 自動ACK存在的問題
修改消費者，添加異常，如下：
生產者不做任何修改，直接運行，消息發送成功：
運行消費者，程序拋出異常。但是消息依然被消費：
管理界面：
1.6 演示手動ACK
修改消費者，把自動改成手動（去掉之前制造的異常）
生產者不變，再次運行：
運行消費者
但是，查看管理界面，發現：
停掉消費者的程序，發現：
這是因為雖然我們設置了手動ACK，但是代碼中並沒有進行消息確認！所以消息並未被真正消費掉。
當我們關掉這個消費者，消息的狀態再次稱為Ready
修改代碼手動ACK：
執行：
消息消費成功！
2. work消息模型
工作隊列或者競爭消費者模式
在第一篇教程中，我們編寫了一個程序，從一個命名隊列中發送並接受消息。在這里，我們將創建一個工作隊列，在多個工作者之間分配耗時任務。
工作隊列，又稱任務隊列。主要思想就是避免執行資源密集型任務時，必須等待它執行完成。相反我們稍后完成任務，我們將任務封裝為消息並將其發送到隊列。 在后台運行的工作進程將獲取任務並最終執行作業。當你運行許多消費者時，任務將在他們之間共享，但是一個消息只能被一個消費者獲取。
這個概念在Web應用程序中特別有用，因為在短的HTTP請求窗口中無法處理復雜的任務。
接下來我們來模擬這個流程：
P：生產者：任務的發布者
C1：消費者，領取任務並且完成任務，假設完成速度較快
C2：消費者2：領取任務並完成任務，假設完成速度慢
面試題：避免消息堆積？
1）采用workqueue，多個消費者監聽同一隊列。
2）接收到消息以后，而是通過線程池，異步消費。
2.1 生產者
生產者與案例1中的幾乎一樣：
package com.example.demorabbitmq.work;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
// 生產者
public class Send {
private final static String QUEUE_NAME = "test_work_queue";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 循環發布任務
for (int i = 0; i < 50; i++) {
// 消息內容
String message = "task .. " + i;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
Thread.sleep(i * 2);
}
// 關閉通道和連接
channel.close();
connection.close();
}
}
不過這里我們是循環發送50條消息。
2.2 消費者1
package com.example.demorabbitmq.work;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
// 消費者1
public class Recv {
private final static String QUEUE_NAME = "test_work_queue";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
final Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 設置每個消費者同時只能處理一條消息
// channel.basicQos(1);
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者1] received : " + msg + "!");
try {
// 模擬完成任務的耗時：1000ms
Thread.sleep(1000);
} catch (InterruptedException e) {
}
// 手動ACK
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
// 監聽隊列。
channel.basicConsume(QUEUE_NAME, false, consumer);
}
}
2.3 消費者2
package com.example.demorabbitmq.work;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
//消費者2
public class Recv2 {
private final static String QUEUE_NAME = "test_work_queue";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
final Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 設置每個消費者同時只能處理一條消息
// channel.basicQos(1);
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者2] received : " + msg + "!");
// 手動ACK
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
// 監聽隊列。
channel.basicConsume(QUEUE_NAME, false, consumer);
}
}
與消費者1基本類似，就是沒有設置消費耗時時間。
這里是模擬有些消費者快，有些比較慢。
接下來，兩個消費者一同啟動，然后發送50條消息：
可以發現，兩個消費者各自消費了25條消息，而且各不相同，這就實現了任務的分發。
2.4 能者多勞
剛才的實現有問題嗎？
- 消費者1比消費者2的效率要低，一次任務的耗時較長
- 然而兩人最終消費的消息數量是一樣的
- 消費者2大量時間處於空閑狀態，消費者1一直忙碌
現在的狀態屬於是把任務平均分配，正確的做法應該是消費越快的人，消費的越多。
怎么實現呢？
我們可以使用basicQos方法和prefetchCount = 1設置。 這告訴RabbitMQ一次不要向工作人員發送多於一條消息。 或者換句話說，不要向工作人員發送新消息，直到它處理並確認了前一個消息。 相反，它會將其分派給不是仍然忙碌的下一個工作人員。
channel.basicQos(1);
再次測試：
3. 訂閱模型分類
在之前的模式中，我們創建了一個工作隊列。 工作隊列背后的假設是：每個任務只被傳遞給一個工作人員。 在這一部分，我們將做一些完全不同的事情 - 我們將會傳遞一個信息給多個消費者。 這種模式被稱為“發布/訂閱”。
訂閱模型示意圖：
解讀：
1、1個生產者，多個消費者
2、每一個消費者都有自己的一個隊列
3、生產者沒有將消息直接發送到隊列，而是發送到了交換機
4、每個隊列都要綁定到交換機
5、生產者發送的消息，經過交換機到達隊列，實現一個消息被多個消費者獲取的目的
X（Exchanges）：交換機一方面：接收生產者發送的消息。另一方面：知道如何處理消息，例如遞交給某個特別隊列、遞交給所有隊列、或是將消息丟棄。到底如何操作，取決於Exchange的類型。
Exchange類型有以下幾種：
Fanout：廣播，將消息交給所有綁定到交換機的隊列
Direct：定向，把消息交給符合指定routing key 的隊列
Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列
Exchange（交換機）只負責轉發消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規則的隊列，那么消息會丟失！
4. 訂閱模型-Fanout
Fanout，也稱為廣播。
流程圖：
在廣播模式下，消息發送流程是這樣的：
- 1） 可以有多個消費者
- 2） 每個**消費者有自己的queue**（隊列）
- 3） 每個**隊列都要綁定到Exchange**（交換機）
- 4） **生產者發送的消息，只能發送到交換機**，交換機來決定要發給哪個隊列，生產者無法決定。
- 5） 交換機把消息發送給綁定過的所有隊列
- 6） 隊列的消費者都能拿到消息。實現一條消息被多個消費者消費
4.1 生產者
兩個變化：
- 1） 聲明Exchange，不再聲明Queue
- 2） 發送消息到Exchange，不再發送到Queue
package com.example.demorabbitmq.fanout;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Send {
private final static String EXCHANGE_NAME = "fanout_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明exchange，指定類型為fanout
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 消息內容
String message = "Hello everyone";
// 發布消息到Exchange
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println(" [生產者] Sent '" + message + "'");
channel.close();
connection.close();
}
}
4.2 消費者1
package com.example.demorabbitmq.fanout;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
//消費者1
public class Recv {
private final static String QUEUE_NAME = "fanout_exchange_queue_1";
private final static String EXCHANGE_NAME = "fanout_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者1] received : " + msg + "!");
}
};
// 監聽隊列，自動返回完成
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
要注意代碼中：隊列需要和交換機綁定
4.3 消費者2
package com.example.demorabbitmq.fanout;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
// 消費者2
public class Recv2 {
private final static String QUEUE_NAME = "fanout_exchange_queue_2";
private final static String EXCHANGE_NAME = "fanout_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者2] received : " + msg + "!");
}
};
// 監聽隊列，手動返回完成
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
4.4 測試
我們運行兩個消費者，然后發送1條消息：
5. 訂閱模型-Direct
有選擇性的接收消息
在訂閱模式中，生產者發布消息，所有消費者都可以獲取所有消息。
在路由模式中，我們將添加一個功能 - 我們將只能訂閱一部分消息。 例如，我們只能將重要的錯誤消息引導到日志文件（以節省磁盤空間），同時仍然能夠在控制台上打印所有日志消息。
但是，在某些場景下，我們希望不同的消息被不同的隊列消費。這時就要用到Direct類型的Exchange。
在Direct模型下，隊列與交換機的綁定，不能是任意綁定了，而是要指定一個RoutingKey（路由key）
消息的發送方在向Exchange發送消息時，也必須指定消息的routing key。
P：生產者，向Exchange發送消息，發送消息時，會指定一個routing key。
X：Exchange（交換機），接收生產者的消息，然后把消息遞交給 與routing key完全匹配的隊列
C1：消費者，其所在隊列指定了需要routing key 為 error 的消息
C2：消費者，其所在隊列指定了需要routing key 為 info、error、warning 的消息
5.1 生產者
此處我們模擬商品的增刪改，發送消息的RoutingKey分別是：insert、update、delete
package com.example.demorabbitmq.direct;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
* 生產者，模擬為商品服務
*/
public class Send {
private final static String EXCHANGE_NAME = "direct_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明exchange，指定類型為direct
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
// 消息內容
// String message = "商品增加了， id = 1001";
// String message = "商品更新了， id = 1002";
String message = "商品刪除了， id = 1003";
// 發送消息，並且指定routing key 為：insert ,代表新增商品
// channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
// channel.basicPublish(EXCHANGE_NAME, "update", null, message.getBytes());
channel.basicPublish(EXCHANGE_NAME, "delete", null, message.getBytes());
System.out.println(" [商品服務：] Sent '" + message + "'");
channel.close();
connection.close();
}
}
5.2 消費者1
我們此處假設消費者1只接收兩種類型的消息：更新商品和刪除商品。
package com.example.demorabbitmq.direct;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* 消費者1
*/
public class Recv {
private final static String QUEUE_NAME = "direct_exchange_queue_1";
private final static String EXCHANGE_NAME = "direct_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機，同時指定需要訂閱的routing key。假設此處需要update和delete消息
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者1] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
5.3 消費者2
我們此處假設消費者2接收所有類型的消息：新增商品，更新商品和刪除商品。
package com.example.demorabbitmq.direct;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* 消費者2
*/
public class Recv2 {
private final static String QUEUE_NAME = "direct_exchange_queue_2";
private final static String EXCHANGE_NAME = "direct_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者2] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
5.4 測試
我們分別發送增、刪、改的RoutingKey，發現結果：
6. 訂閱模型-Topic
Topic類型的Exchange與Direct相比，都是可以根據RoutingKey把消息路由到不同的隊列。只不過Topic類型Exchange可以讓隊列在綁定Routing key 的時候使用通配符！
Routingkey 一般都是有一個或多個單詞組成，多個單詞之間以”.”分割，例如： item.insert
通配符規則：
`#`：匹配一個或多個詞
`*`：匹配不多不少恰好1個詞
舉例：
`audit.#`：能夠匹配`audit.irs.corporate` 或者 `audit.irs`
`audit.*`：只能匹配`audit.irs`
在這個例子中，我們將發送所有描述動物的消息。消息將使用由三個字（兩個點）組成的routing key發送。路由關鍵字中的第一個單詞將描述速度，第二個顏色和第三個種類：“
我們創建了三個綁定：Q1綁定了綁定鍵"* .orange."，Q2綁定了".*.rabbit"和"lazy.＃"。
Q1匹配所有的橙色動物。
Q2匹配關於兔子以及懶惰動物的消息。
6.1 生產者
使用topic類型的Exchange，發送消息的routing key有3種： item.isnert、item.update、item.delete：
package com.example.demorabbitmq.topic;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
* 生產者，模擬為商品服務
*/
public class Send {
private final static String EXCHANGE_NAME = "topic_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明exchange，指定類型為topic
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
// 消息內容
String message = "新增商品 : id = 1001";
// 發送消息，並且指定routing key 為：insert ,代表新增商品
channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes());
System.out.println(" [商品服務：] Sent '" + message + "'");
channel.close();
connection.close();
}
}
6.2 消費者1
我們此處假設消費者1只接收兩種類型的消息：更新商品和刪除商品
package com.example.demorabbitmq.topic;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* 消費者2
*/
public class Recv2 {
private final static String QUEUE_NAME = "topic_exchange_queue_2";
private final static String EXCHANGE_NAME = "topic_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者2] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
6.3 消費者2
我們此處假設消費者2接收所有類型的消息：新增商品，更新商品和刪除商品。
package com.example.demorabbitmq.topic;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* 消費者2
*/
public class Recv2 {
private final static String QUEUE_NAME = "topic_exchange_queue_2";
private final static String EXCHANGE_NAME = "topic_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者2] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
6.4 執行測試
7. 持久化
如何避免消息丟失？
1） 消費者的ACK機制。可以防止消費者丟失消息。
2） 但是，如果在消費者消費之前，MQ就宕機了，消息就沒了。
是可以將消息進行持久化呢？
要將消息持久化，前提是：隊列、Exchange都持久化
7.1 交換機持久化
7.2 隊列持久化
7.3 消息持久化
7.4 以topic demo為例測試
此時的exchange topic_exchange_test並不包含D 說明不能持久
即使代碼根據上面提示進行修改，運行依然會報錯，因為exchange已經存在，需要手動刪除,queue同理也需要手動刪除
此時再次瀏覽
已經修改為D
此時停止消費者12程序，再次運行生產者
會看到,queue中有一條待處理的消息
此時重啟rabbitmq
重新在瀏覽器端查看，可以發現消息依然存在
生產者code
package com.example.demorabbitmq.topic;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
/**
* 生產者，模擬為商品服務
*/
public class Send {
private final static String EXCHANGE_NAME = "topic_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明exchange，指定類型為topic
channel.exchangeDeclare(EXCHANGE_NAME, "topic",true);
// 消息內容
String message = "新增商品 : id = 1001";
// 發送消息，並且指定routing key 為：insert ,代表新增商品
channel.basicPublish(EXCHANGE_NAME, "item.insert", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
System.out.println(" [商品服務：] Sent '" + message + "'");
channel.close();
connection.close();
}
}
消費者1 code
package com.example.demorabbitmq.topic;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* 消費者1
*/
public class Recv {
private final static String QUEUE_NAME = "topic_exchange_queue_1";
private final static String EXCHANGE_NAME = "topic_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 綁定隊列到交換機，同時指定需要訂閱的routing key。需要 update、delete
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者1] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
消費者2 code
package com.example.demorabbitmq.topic;
import java.io.IOException;
import com.example.demorabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* 消費者2
*/
public class Recv2 {
private final static String QUEUE_NAME = "topic_exchange_queue_2";
private final static String EXCHANGE_NAME = "topic_exchange_test";
public static void main(String[] argv) throws Exception {
// 獲取到連接
Connection connection = ConnectionUtil.getConnection();
// 獲取通道
Channel channel = connection.createChannel();
// 聲明隊列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*");
// 定義隊列的消費者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 獲取消息，並且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
byte[] body) throws IOException {
// body 即消息體
String msg = new String(body);
System.out.println(" [消費者2] received : " + msg + "!");
}
};
// 監聽隊列，自動ACK
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}