RabbitMQ 从入门到精通 （一）
生活随笔
收集整理的這篇文章主要介紹了
RabbitMQ 从入门到精通 （一）
小編覺得挺不錯的,現在分享給大家,幫大家做個參考.
目錄
- 1. 初識RabbitMQ
- 2. AMQP
- 3.RabbitMQ的極速入門
- 4. Exchange(交換機)詳解
- 4.1 Direct Exchange
- 4.2 Topic Exchange
- 4.3 Fanout Exchange
- 5. Message 消息
1. 初識RabbitMQ
RabbitMQ 是一個開源的消息代理和隊列服務器，用來通過普通協議在完全不同的應用之間共享數據，RabbitMQ是使用 Erlang語言來編寫的，并且RabbitMQ是基于AMQP協議的
RabbitMQ的優點:
- 開源、性能優秀、穩定性保障
- 提供可靠性消息投遞模式(confirm)、返回模式(return)
- 與SpringAMQP完美的整合、API豐富
- 集群模式豐富，表達式配置，HA模式，鏡像隊列模型
- 保證數據不丟失的前提下做到高可靠性、可用性
RabbitMQ官網
RabbitMQ的整體架構:
?
RabbitMQ的消息流轉:
?
?
2. AMQP
AMQP全稱: Advanced Message Queuing Protocol
AMQP翻譯: 高級消息隊列協議
AMQP定義: 是具有現代特征的二進制協議。是一個提供統一消息服務的應用層標準高級消息隊列協議，是應用層協議的一個開放標準，為面向消息的中間件設計
?
?
AMQP核心概念：
- Server:又稱Broker，接受客戶端的連接，實現AMQP實體服務
- Connection:連接，應用程序與Broker的網絡連接
- Channel:網絡信道，幾乎所有的操作都在Channel中進行，Channel是進行消息讀寫的通道。客戶端可建立多個Channel，每個Channel代表一個會話任務
- Message:消息，服務器和應用程序之間傳送的數據，由Properties和Body組成。Properties可以對消息進行修飾，比如消息的優先級、延遲等高級特性；Body則是消息體的內容
- Virtual host:虛擬地址，用于進行邏輯隔離，最上層的消息路由。同一個Virtual Host里面不能有相同名稱的Exchange或Queue
- Exchange:交換機，接收消息，根據路由鍵轉發消息到綁定的隊列
- Binding:Exchange和Queue之間的虛擬連接,binding中可以包含routing key
- Routing key:一個路由規則，虛擬機可用它確定如何路由一個特定消息
- Queue:也稱為Message Queue，消息隊列，保存消息并將它們轉發給消費者
?
?
3.RabbitMQ的極速入門
后臺啟動: ./rabbitmq start &
關閉: ./rabbitmqctl stop
節點狀態: ./rabbitmqctl status
管控臺: http://ip:15672
?
?
RabbitMQ生產消費快速入門:
環境： springboot+jdk1.7+rabbitmq3.6.5 (Maven依賴配置)
<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>1.5.9.RELEASE</version></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>3.6.5</version></dependency></dependencies>?
public class Procuder {public static void main(String[] args) throws Exception {//1.創建一個ConnectionFactory 并進行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setHandshakeTimeout(20000);//2.通過連接工廠創建連接Connection connection = connectionFactory.newConnection();//3.通過Connection 創建一個 ChannelChannel channel = connection.createChannel();/*** basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)* exchange:指定交換機 不指定 則默認 （AMQP default交換機） 通過routingkey進行匹配 * props 消息屬性* body 消息體*///4.通過Channel發送數據for(int i = 0; i < 5; i++){System.out.println("生產消息:" + i);String msg = "Hello RabbitMQ" + i;channel.basicPublish("", "test", null, msg.getBytes());}//5.記得關閉相關的連接channel.close();connection.close();} }?
public class Consumer {public static void main(String[] args) throws Exception{//1.創建一個ConnectionFactory 并進行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setHandshakeTimeout(20000);//2.通過連接工廠創建連接Connection connection = connectionFactory.newConnection();//3.通過Connection 創建一個 ChannelChannel channel = connection.createChannel();//4. 聲明創建一個隊列String queueName = "test";/*** durable 是否持久化* exclusive 獨占的 相當于加了一把鎖*/channel.queueDeclare(queueName,true,false,false,null);//5.創建消費者QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//6.設置channel/*** ACK: 當一條消息從生產端發到消費端，消費端接收到消息后會馬上回送一個ACK信息給broker,告訴它這條消息收到了* autoack: * true 自動簽收 當消費者一收到消息就表示消費者收到了消息，消費者收到了消息就會立即從隊列中刪除。* false 手動簽收 當消費者收到消息在合適的時候來顯示的進行確認，說我已經接收到了該消息了，RabbitMQ可以從隊列中刪除該消息了* */channel.basicConsume(queueName, true, queueingConsumer);//7.獲取消息while(true){Delivery delivery = queueingConsumer.nextDelivery();String msg = new String(delivery.getBody());System.err.println("消費端:" + msg);//Envelope envelope = delivery.getEnvelope();}} }?
4. Exchange(交換機)詳解
Exchange: 接收消息，并根據路由鍵轉發消息所綁定的隊列
?
交換機屬性:
- Name: 交換機名稱
- Type: 交換機類型 diect、topic、fanout、headers
- Durability: 是否需要持久化，true為持久化
- AutoDelete: 當最后一個綁定到Exchange的隊列刪除后，自動刪除該Exchange
- Internal: 當前Exchange是否用于RabbitMQ內部使用，默認為false (百分之99的情況默認為false 除非對Erlang語言較了解，做一些擴展)
- Arguments: 擴展參數, 用于擴展AMQP協議可自定化使用
?
4.1 Direct Exchange
所有發送到Direct Exchange的消息被轉發到RouteKey指定的Queue
注意：Direct模式可以使用RabbitMQ自帶的Exchange: default Exchange，所以不需要將Exchange進行任何綁定(binding)操作，消息傳遞時，RoutingKey必須完全匹配才會被隊列接收，否則該消息會被拋棄
?
public class ProducerDirectExchange {public static void main(String[] args) throws Exception {//1.創建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2.創建ConnectionConnection connection = connectionFactory.newConnection();//3.創建ChannelChannel channel = connection.createChannel();//4.聲明String exchangeName = "test_direct_exchange";String routingKey = "test.direct";//5.發送String msg = "Hello World RabbitMQ4 Direct Exchange Message";channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());} }?
public class ConsumerDirectExchange {public static void main(String[] args) throws Exception{ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setHandshakeTimeout(20000);connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//聲明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,Consumerchannel.basicConsume(queueName, true, consumer);//循環獲取消息while(true){//獲取消息,如果沒有消息,這一步將會一直阻塞Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println("收到消息:" + msg);}} }?
4.2 Topic Exchange
所有發送到Topic Exchange的消息被轉發到所有關心RouteKey中指定Topic的Queue上
Exchange將RouteKey和某Topic進行模糊匹配，此時隊列需要綁定一個Topic
注意：可以使用通配符進行匹配
符號 # 匹配一個或多個詞
符號 * 匹配不多不少一個詞
例如: "log.#" 能夠匹配到 “log.info.oa”
? "log.*" 只會匹配到 "log.err"
public class ProducerTopicExchange {public static void main(String[] args) throws Exception {//1.創建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setHandshakeTimeout(20000);//2.創建ConnectionConnection connection = connectionFactory.newConnection();//3.創建ChannelChannel 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 RabbitMQ4 Direct Exchange Message";channel.basicPublish(exchangeName, routingKey1, null, msg.getBytes());channel.basicPublish(exchangeName, routingKey2, null, msg.getBytes());channel.basicPublish(exchangeName, routingKey3, null, msg.getBytes());} }?
public class ConsumerTopicExchange {public static void main(String[] args) throws Exception{ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setHandshakeTimeout(20000);connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//聲明String exchangeName = "test_topic_exchange";String exchangeType = "topic";String queueName = "test_topic_queue";String routingKey = "user.#";//表示聲明了一個交換機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,Consumerchannel.basicConsume(queueName, true, consumer);//循環獲取消息while(true){//獲取消息,如果沒有消息,這一步將會一直阻塞Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println("收到消息:" + msg);}} }?
4.3 Fanout Exchange
不處理路由鍵，只需要簡單的將隊列綁定到交換機上
發送到交換機的消息都會被轉發到與該交換機綁定的所有隊列上
所以Fanout交換機轉發消息是最快的
?
public class ProducerFanoutExchange {public static void main(String[] args) throws Exception {//1.創建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setHandshakeTimeout(20000);//2.創建ConnectionConnection connection = connectionFactory.newConnection();//3.創建ChannelChannel channel = connection.createChannel();//4.聲明String exchangeName = "test_fanout_exchange";//5.發送for(int i = 0; i < 10 ; i++){String msg = "Hello World RabbitMQ4 Direct Exchange Message";channel.basicPublish(exchangeName, "", null, msg.getBytes());}channel.close();connection.close();} }?
public class ConsumerFanoutExchange {public static void main(String[] args) throws Exception{ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setHandshakeTimeout(20000);connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//聲明String exchangeName = "test_fanout_exchange";String exchangeType = "fanout";String queueName = "test_topic_queue";//無需指定路由key 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,Consumerchannel.basicConsume(queueName, true, consumer);//循環獲取消息while(true){//獲取消息,如果沒有消息,這一步將會一直阻塞Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println("收到消息:" + msg);}} }?
5. Message 消息
服務器與應用程序之間傳遞的數據，本質上就是一段數據，由Properties和Body組成
常用屬性：delivery mode、headers (自定義屬性)
其他屬性：content_type、content_encoding、priority、expiration
消息的properties屬性用法示例:
public class Procuder {public static void main(String[] args) throws Exception {//1.創建一個ConnectionFactory 并進行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setHandshakeTimeout(20000);//2.通過連接工廠創建連接Connection connection = connectionFactory.newConnection();//3.通過Connection 創建一個 ChannelChannel channel = connection.createChannel();Map<String,Object> headers = new HashMap<>();headers.put("my1", "111");headers.put("my2", "222");//10秒不消費 消息過期移除消息隊列AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().deliveryMode(2).contentEncoding("utf-8").expiration("10000").headers(headers).build();//4.通過Channel發送數據for(int i = 0; i < 5; i++){System.out.println("生產消息:" + i);String msg = "Hello RabbitMQ" + i;channel.basicPublish("", "test", properties, msg.getBytes());}//5.記得關閉相關的連接channel.close();connection.close();} }?
public class Consumer {public static void main(String[] args) throws Exception{//1.創建一個ConnectionFactory 并進行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.244.11");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setHandshakeTimeout(20000);//2.通過連接工廠創建連接Connection connection = connectionFactory.newConnection();//3.通過Connection 創建一個 ChannelChannel channel = connection.createChannel();//4. 聲明創建一個隊列String queueName = "test";channel.queueDeclare(queueName,true,false,false,null);//5.創建消費者QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//6.設置channelchannel.basicConsume(queueName, true, queueingConsumer);//7.獲取消息while(true){Delivery delivery = queueingConsumer.nextDelivery();String msg = new String(delivery.getBody());System.err.println("消費端:" + msg);Map<String, Object> headers = delivery.getProperties().getHeaders();System.err.println("headers value:" + headers.get("my1"));}} }轉載于:https://www.cnblogs.com/dwlovelife/p/10982735.html
總結
以上是生活随笔為你收集整理的RabbitMQ 从入门到精通 （一）的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 流程控制元素
- 下一篇： 有进度条圆周率Π计算