作者：海向
出處：cnblogs.com/haixiang/p/10853467.html
RabbitMQ 特點
RabbitMQ 相較于其他訊息佇列，有一系列防止訊息丟失的措施，擁有強悍的高可用性能，它的吞吐量可能沒有其他訊息佇列大，但是其訊息的保障性出類拔萃，被廣泛用于金融類業務，
AMQP 協議
AMQP: Advanced Message Queuing Protocol 高級訊息佇列協議
AMQP定義:是具有現代特征的二進制協議，是一個提供統一訊息服務的應用層標準高級訊息佇列協議，是應用層協議的一個開放標準，為面向訊息的中間件設計，
Erlang語言最初在于交換機領域的架構模式，這樣使得RabbitMQ在Broker之間進行資料互動的性能是非常優秀的，Erlang的優點: Erlang有著和原生Socket一樣的延遲，
RabbitMQ是一個開源的訊息代理和佇列服務器，用來通過普通協議在完全不同的應用之間共享資料, RabbitMQ是使用Erlang語言來撰寫的，并且RabbitMQ是基于AMQP協議的，關注公眾號Java技術堆疊獲取系列RabbitMQ教程，
RabbitMQ 訊息傳遞機制
生產者發送訊息到指定的 Exchange，Exchange 依據自身的型別(direct、topic等)，根據 routing key 將訊息發送給 0 - n 個 佇列，佇列再將訊息轉發給了消費者，
Server: 又稱Broker, 接受客戶端的連接，實作AMQP物體服務，這里指RabbitMQ 服務器
Connection: 連接，應用程式與Broker的網路連接，
Channel: 網路信道，幾乎所有的操作都在 Channel 中進行，Channel是進行訊息讀寫的通道，客戶端可建立多個Channel:，每個Channel代表一個會話任務，
Virtual host: 虛似地址，用于迸行邏輯隔離，是最上層的訊息路由，一個 Virtual Host 里面可以有若干個 Exchange和 Queue ，同一個 VirtualHost 里面不能有相同名稱的 Exchange 或 Queue，權限控制的最小粒度是Virtual Host，
Binding: Exchange 和 Queue 之間的虛擬連接，binding 中可以包含 routing key，
Routing key: 一 個路由規則，虛擬機可用它來確定如何路由一個特定訊息，即交換機系結到 Queue 的鍵，
Queue: 也稱為Message Queue,訊息佇列，保存訊息并將它們轉發給消費者，
Message
訊息，服務器和應用程式之間傳送的資料，由 Properties 和 Body 組成，Properties 可以對訊息進行修飾，比如訊息的優先級、延遲等高級特性;，Body 則就 是訊息體內容，
properties 中我們可以設定訊息過期時間以及是否持久化等，也可以傳入自定義的map屬性，這些在消費端也都可以獲取到，
生產者
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;
public class MessageProducer {
public static void main(String[] args) throws Exception {
//1. 創建一個 ConnectionFactory 并進行設定
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
//2. 通過連接工廠來創建連接
Connection connection = factory.newConnection();
//3. 通過 Connection 來創建 Channel
Channel channel = connection.createChannel();
//4. 宣告 使用默認交換機 以佇列名作為 routing key
String queueName = "msg_queue";
/**
* deliverMode 設定為 2 的時候代表持久化訊息
* expiration 意思是設定訊息的有效期，超過10秒沒有被消費者接收后會被自動洗掉
* headers 自定義的一些屬性
* */
//5. 發送
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("myhead1", "111");
headers.put("myhead2", "222");
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
.deliveryMode(2)
.contentEncoding("UTF-8")
.expiration("100000")
.headers(headers)
.build();
String msg = "test message";
channel.basicPublish("", queueName, properties, msg.getBytes());
System.out.println("Send message : " + msg);
//6. 關閉連接
channel.close();
connection.close();
}
}
消費者
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.Map;
public class MessageConsumer {
public static void main(String[] args) throws Exception{
//1. 創建一個 ConnectionFactory 并進行設定
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
factory.setAutomaticRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(3000);
//2. 通過連接工廠來創建連接
Connection connection = factory.newConnection();
//3. 通過 Connection 來創建 Channel
Channel channel = connection.createChannel();
//4. 宣告
String queueName = "msg_queue";
channel.queueDeclare(queueName, false, false, false, null);
//5. 創建消費者并接收訊息
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
Map<String, Object> headers = properties.getHeaders();
System.out.println("head: " + headers.get("myhead1"));
System.out.println(" [x] Received '" + message + "'");
System.out.println("expiration : "+ properties.getExpiration());
}
};
//6. 設定 Channel 消費者系結佇列
channel.basicConsume(queueName, true, consumer);
}
}
Send message : test message
head: 111
[x] Received 'test message'
100000
Exchange
1. 簡介
Exchange 就是交換機，接收訊息，根據路由鍵轉發訊息到系結的佇列，有很多的 Message 進入到 Exchange 中，Exchange 根據 Routing key 將 Message 分發到不同的 Queue 中，
2. 型別
RabbitMQ中的 Exchange 有多種型別，型別不同，Message 的分發機制不同，如下：
-
fanout：廣播模式，這種型別的 Exchange 會將 Message 分發到系結到該 Exchange 的所有 Queue，
-
direct：這種型別的 Exchange 會根據 Routing key（精確匹配，將Message分發到指定的Queue，
-
Topic：這種型別的 Exchange 會根據 Routing key（模糊匹配，將Message分發到指定的Queue，
-
headers: 主題交換機有點相似，但是不同于主題交換機的路由是基于路由鍵，頭交換機的路由值基于訊息的header資料，主題交換機路由鍵只有是字串,而頭交換機可以是整型和哈希值 .
3. 屬性
/**
* Declare an exchange, via an interface that allows the complete set of
* arguments.
* @see com.rabbitmq.client.AMQP.Exchange.Declare
* @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
* @param exchange the name of the exchange
* @param type the exchange type
* @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
* @param autoDelete true if the server should delete the exchange when it is no longer in use
* @param internal true if the exchange is internal, i.e. can't be directly
* published to by a client.
* @param arguments other properties (construction arguments) for the exchange
* @return a declaration-confirm method to indicate the exchange was successfully declared
* @throws java.io.IOException if an error is encountered
*/
Exchange.DeclareOk exchangeDeclare(String exchange,
String type,boolean durable,
boolean autoDelete,boolean internal,
Map<String, Object> arguments) throws IOException;
-
Name: 交換機名稱
-
Type: 交換機型別direct、topic、 fanout、 headers
-
Durability: 是否需要持久化，true為持久化
-
Auto Delete: 當最后一個系結到Exchange. 上的佇列洗掉后，自動洗掉該Exchange
-
Internal: 當前Exchange是否用于RabbitMQ內部使用，默認為False
-
Arguments: 擴展引數，用于擴展AMQP協議自制定化使用
近期熱文推薦：
1.終于靠開源專案弄到 IntelliJ IDEA 激活碼了，真香！
2.我用 Java 8 寫了一段邏輯，同事直呼看不懂，你試試看，，
3.吊打 Tomcat ，Undertow 性能很炸！！
4.國人開源了一款超好用的 Redis 客戶端，真香！！
5.《Java開發手冊（嵩山版）》最新發布，速速下載！
覺得不錯，別忘了隨手點贊+轉發哦！
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/104902.html
標籤：Java
上一篇：學習第33天