RabbitMQ工作模式-Publish/Subscribe發(fā)布與訂閱模式
作者：海洋的漁夫
Exchange（交換機）只負責轉(zhuǎn)發(fā)消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規(guī)則的隊列，那么消息會丟失！
訂閱模式類型
訂閱模式示例圖：
前面2個案例中，只有3個角色：
- P：生產(chǎn)者，也就是要發(fā)送消息的程序
- C：消費者：消息的接受者，會一直等待消息到來。
- queue：消息隊列，圖中紅色部分
而在訂閱模型中，多了一個exchange角色，而且過程略有變化：
- P：生產(chǎn)者，也就是要發(fā)送消息的程序，但是不再發(fā)送到隊列中，而是發(fā)給X（交換機）
- C：消費者，消息的接受者，會一直等待消息到來。
- Queue：消息隊列，接收消息、緩存消息。
- Exchange：交換機，圖中的X。一方面，接收生產(chǎn)者發(fā)送的消息。另一方面，知道如何處理消息，例如遞交給某個特別隊列、遞交給所有隊列、或是將消息丟棄。到底如何操作，取決于Exchange的類型。Exchange有常見以下3種類型：
- Fanout：廣播，將消息交給所有綁定到交換機的隊列
- Direct：定向，把消息交給符合指定routing key 的隊列
- Topic：通配符，把消息交給符合routing pattern（路由模式） 的隊列
Exchange（交換機）只負責轉(zhuǎn)發(fā)消息，不具備存儲消息的能力，因此如果沒有任何隊列與Exchange綁定，或者沒有符合路由規(guī)則的隊列，那么消息會丟失！
Publish/Subscribe發(fā)布與訂閱模式
1、模式說明
發(fā)布訂閱模式：
每個消費者監(jiān)聽自己的隊列。
生產(chǎn)者將消息發(fā)給broker，由交換機將消息轉(zhuǎn)發(fā)到綁定此交換機的每個隊列，每個綁定交換機的隊列都將接收 到消息
2、案例
（1）生產(chǎn)者
package com.lijw.producer;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author Aron.li
* @date 2022/3/3 8:16
*/
public class Producer_PubSub {
public static void main(String[] args) throws IOException, TimeoutException {
//1.創(chuàng)建連接工廠
ConnectionFactory factory = new ConnectionFactory();
//2. 設(shè)置參數(shù)
factory.setHost("127.0.0.1"); // ip 默認值 localhost
factory.setPort(5672); //端口 默認值 5672
factory.setVirtualHost("/test"); //虛擬機 默認值 /
factory.setUsername("libai"); // 用戶名 默認 guest
factory.setPassword("libai"); //密碼 默認值 guest
//3. 創(chuàng)建連接 Connection
Connection connection = factory.newConnection();
//4. 創(chuàng)建Channel
Channel channel = connection.createChannel();
//5. 創(chuàng)建交換機
/*
exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
參數(shù)：
1. exchange:交換機名稱
2. type:交換機類型
DIRECT("direct")：定向
FANOUT("fanout")：扇形（廣播），發(fā)送消息到每一個與之綁定隊列。
TOPIC("topic") 通配符的方式
HEADERS("headers") 參數(shù)匹配
3. durable:是否持久化
4. autoDelete:自動刪除
5. internal：內(nèi)部使用。 一般false
6. arguments：參數(shù)
*/
String exchangeName = "test_fanout";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT, true, false, false, null);
//6. 創(chuàng)建隊列
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
channel.queueDeclare(queue1Name, true, false, false, null);
channel.queueDeclare(queue2Name, true, false, false, null);
// 7. 綁定隊列和交換機
/*
queueBind(String queue, String exchange, String routingKey)
參數(shù)：
1. queue：隊列名稱
2. exchange：交換機名稱
3. routingKey：路由鍵，綁定規(guī)則
如果交換機的類型為fanout ，routingKey設(shè)置為""
*/
channel.queueBind(queue1Name, exchangeName, "");
channel.queueBind(queue2Name, exchangeName, "");
//8. 發(fā)送消息至交換機，由交換機分發(fā)消息
String body = "日志信息: 肥仔白調(diào)用了findAll方法...日志級別: INFO....";
channel.basicPublish(exchangeName, "", null, body.getBytes());
//9. 釋放資源
channel.close();
connection.close();
}
}執(zhí)行生產(chǎn)者，我們可以查看一下創(chuàng)建的 交換機 以及 隊列信息：
下面再來看看隊列，如下：
下面我們繼續(xù)來寫兩個消費者接收消息。
（2）消費者1：讀取隊列1的消息
package com.lijw.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author Aron.li
* @date 2022/3/2 16:16
*/
public class Consumer_PubSub1 {
//定義接收隊列的名稱
final static String queueName = "test_fanout_queue1";
public static void main(String[] args) throws IOException, TimeoutException {
//1.創(chuàng)建連接工廠
ConnectionFactory factory = new ConnectionFactory();
//2. 設(shè)置參數(shù)
factory.setHost("127.0.0.1"); // ip 默認值 localhost
factory.setPort(5672); //端口 默認值 5672
factory.setVirtualHost("/test"); //虛擬機 默認值 /
factory.setUsername("libai"); // 用戶名 默認 guest
factory.setPassword("libai"); //密碼 默認值 guest
//3. 創(chuàng)建連接 Connection
Connection connection = factory.newConnection();
//4. 創(chuàng)建Channel
Channel channel = connection.createChannel();
//5. 創(chuàng)建隊列Queue
/*
queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
參數(shù)：
1. queue：隊列名稱
2. durable:是否持久化，當mq重啟之后，還在
3. exclusive：
* 是否獨占。只能有一個消費者監(jiān)聽這隊列
* 當Connection關(guān)閉時，是否刪除隊列
4. autoDelete:是否自動刪除。當沒有Consumer時，自動刪除掉
5. arguments：參數(shù)。
*/
channel.queueDeclare(queueName, true, false, false, null);
/*
basicConsume(String queue, boolean autoAck, Consumer callback)
參數(shù)：
1. queue：隊列名稱
2. autoAck：是否自動確認
3. callback：回調(diào)對象
*/
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
/*
回調(diào)方法，當收到消息后，會自動執(zhí)行該方法
1. consumerTag：標識
2. envelope：獲取一些信息，交換機，路由key...
3. properties:配置信息
4. body：數(shù)據(jù)
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("接收隊列的數(shù)據(jù) body: " + new String(body));
}
};
channel.basicConsume(queueName,true,consumer);
//不需要關(guān)閉資源，因為消費者需要持續(xù)監(jiān)聽隊列信息
}
}（3）消費者2：讀取隊列2的消息
package com.lijw.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author Aron.li
* @date 2022/3/2 16:16
*/
public class Consumer_PubSub2 {
//定義接收隊列的名稱
final static String queueName = "test_fanout_queue2";
public static void main(String[] args) throws IOException, TimeoutException {
//1.創(chuàng)建連接工廠
ConnectionFactory factory = new ConnectionFactory();
//2. 設(shè)置參數(shù)
factory.setHost("127.0.0.1"); // ip 默認值 localhost
factory.setPort(5672); //端口 默認值 5672
factory.setVirtualHost("/test"); //虛擬機 默認值 /
factory.setUsername("libai"); // 用戶名 默認 guest
factory.setPassword("libai"); //密碼 默認值 guest
//3. 創(chuàng)建連接 Connection
Connection connection = factory.newConnection();
//4. 創(chuàng)建Channel
Channel channel = connection.createChannel();
//5. 創(chuàng)建隊列Queue
/*
queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
參數(shù)：
1. queue：隊列名稱
2. durable:是否持久化，當mq重啟之后，還在
3. exclusive：
* 是否獨占。只能有一個消費者監(jiān)聽這隊列
* 當Connection關(guān)閉時，是否刪除隊列
4. autoDelete:是否自動刪除。當沒有Consumer時，自動刪除掉
5. arguments：參數(shù)。
*/
channel.queueDeclare(queueName, true, false, false, null);
/*
basicConsume(String queue, boolean autoAck, Consumer callback)
參數(shù)：
1. queue：隊列名稱
2. autoAck：是否自動確認
3. callback：回調(diào)對象
*/
// 接收消息
Consumer consumer = new DefaultConsumer(channel){
/*
回調(diào)方法，當收到消息后，會自動執(zhí)行該方法
1. consumerTag：標識
2. envelope：獲取一些信息，交換機，路由key...
3. properties:配置信息
4. body：數(shù)據(jù)
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("接收隊列的數(shù)據(jù) body: " + new String(body));
}
};
channel.basicConsume(queueName,true,consumer);
//不需要關(guān)閉資源，因為消費者需要持續(xù)監(jiān)聽隊列信息
}
}3、測試
啟動所有消費者，然后使用生產(chǎn)者發(fā)送消息；在每個消費者對應(yīng)的控制臺可以查看到生產(chǎn)者發(fā)送的所有消息；到達廣播的效果。
- 消費者1接收到的消息：
- 消費者2接收到的消息：
從結(jié)果來看，生產(chǎn)者只需要發(fā)送一條消息，其余的消費者全部收到了消息，達到了廣播的效果。
4、小結(jié)
交換機需要與隊列進行綁定，綁定之后；一個消息可以被多個消費者都收到。
發(fā)布訂閱模式與工作隊列模式的區(qū)別：
- 工作隊列模式不用定義交換機，而發(fā)布/訂閱模式需要定義交換機。
- 發(fā)布/訂閱模式的生產(chǎn)方是面向交換機發(fā)送消息，工作隊列模式的生產(chǎn)方是面向隊列發(fā)送消息(底層使用默認交換機)。
- 發(fā)布/訂閱模式需要設(shè)置隊列和交換機的綁定，工作隊列模式不需要設(shè)置，實際上工作隊列模式會將隊列綁 定到默認的交換機 。
責任編輯：姜華
來源：
今日頭條
