目錄

rabbitmq

術語

代碼

概念

---

rabbitmq

術語

RabbitMQ： 實現了消息隊列協議（AMQP）的開源消息代理軟件及面向消息的中間件。RabbitMQ服務器是用Erlang語言編寫的。

AMQP ：一個提供統一消息服務的應用層標準高級消息隊列協議，是應用層協議的一個開放標準，為面向消息的中間件設計。消息中間件主要用于組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。

生產者：投遞消息的一方，生產者創建消息，然后發布到rabbitmq中。

消費者，就是接收消息的 一方。消費者連接到 RabbitMQ 服務器，并訂閱到隊列上 。 消費者也不需要知道誰是生產者 。

隊列：隊列就像一個超市或者流水線，在RabbitMQ中，生產者創建消息，來到RabbitMQ的隊列，消費者可以從隊列中取出消息進行消費。所有信息可以只存儲在一個隊列中。隊列可以存儲很多的消息，因為它基本上是一個無限制的緩沖區，前提是你的機器有足夠的存儲空間。多個生產者可以將消息發送到同一個隊列中，多個消費者也可以只從同一個隊列接收數據。這就是隊列的特性。

代碼

使用工具是eclipse，創建maven項目 ，可以引入依賴

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

demo目錄

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?

生產者：

package com.ll.mq.hellomq;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory;/*** * @author ll 生產者**/ public class Producer {public final static String QUENE_NAME = "hello";// 定義隊列名稱public static void main(String[] args) {try {// 創建連接工廠ConnectionFactory factory = new ConnectionFactory();// RabbitMQ地址factory.setHost("127.0.0.1");factory.setPort(5672);factory.setUsername("kysc");factory.setPassword("123456");// 創建一個連接Connection connection = factory.newConnection();// 創建一個頻道Channel channel = connection.createChannel();// 聲明一個隊列 在RabbitMQ中，隊列聲明是冪等性的（一個冪等操作的特點是其任意多次執行所產生的影響均與一次執行的影響相同），也就是說，如果不存在，就創建，如果存在，不會對已經存在的隊列產生任何影響。channel.queueDeclare(QUENE_NAME, false, false, false, null);String message = "HELLO MQ!";// 發送消息到隊列中channel.basicPublish("", QUENE_NAME, null, message.getBytes("UTF-8"));System.out.println("Producer [x] Send '" + message + "'");// 關閉頻道和連接channel.close();connection.close();} catch (Exception e) {e.printStackTrace();} }}

消費者

package com.ll.mq.hellomq;import java.io.IOException;import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; /*** * @author ll Consumer **/ public class Consumer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {// 創建連接工廠ConnectionFactory factory = new ConnectionFactory(); // 設置RabbitMQ地址factory.setHost("127.0.0.1");factory.setPort(5672);factory.setUsername("kysc");factory.setPassword("123456"); // 創建一個連接Connection connection = factory.newConnection(); // 創建一個頻道Channel channel = connection.createChannel(); // 聲明要關注的隊列 -- 在RabbitMQ中，隊列聲明是冪等性的（一個冪等操作的特點是其任意多次執行所產生的影響均與一次執行的影響相同），也就是說，如果不存在，就創建，如果存在，不會對已經存在的隊列產生任何影響。channel.queueDeclare(QUEUE_NAME, false, false, false, null);System.out.println("C [*] Waiting for messages. To exit press CTRL+C"); // DefaultConsumer類實現了Consumer接口，通過傳入一個頻道，告訴服務器我們需要那個頻道的消息，如果頻道中有消息，就會執行回調函數handleDeliveryDefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("C [x] Received '" + message + "'");}}; // 自動回復隊列應答 -- RabbitMQ中的消息確認機制，后面章節會詳細講解channel.basicConsume(QUEUE_NAME, true, consumer);} }

接受結果：C [*] Waiting for messages. To exit press CTRL+C
? ? ? ? ? ? ? ? ? C [x] Received 'HELLO MQ!'

概念

ConnectionFactory(連接工廠): 生產Connection的的工廠

Connection(連接):是RabbitMQ的socket的長鏈接，它封裝了socket協議相關部分邏輯

Channel(頻道|信道): 是建立在Connection連接之上的一種輕量級的連接，我們大部分的業務操作是在Channel這個接口中完成的，包括定義隊列的聲明queueDeclare、交換機的聲明exchangeDeclare、隊列的綁定queueBind、發布消息basicPublish、消費消息basicConsume等。

public Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? boolean autoDelete, Map<String, Object> arguments)

queue: 隊列名稱

durable： 是否持久化, 隊列的聲明默認是存放到內存中的，如果rabbitmq重啟會丟失，如果想重啟之后還存在就要使隊列持久化，保存到Erlang自帶的Mnesia數據庫中，當rabbitmq重啟之后會讀取該數據庫

exclusive：是否排外

1 當exclusive = true則設置隊列為排他的。如果一個隊列被聲明為排他隊列，該隊列 僅對首次聲明它的連接（Connection）可見，是該Connection私有的，類似于加鎖，并在連接斷開connection.close()時自動刪除 ；
2 exclusive = false則設置隊列為非排他的，此時不同連接（Connection）的管道Channel可以使用該隊列 ；
"首次" 是指如果某個連接（Connection）已經聲明了排他隊列，其他連接是不允許建立同名的排他隊列的。這個與普通隊列不同：即使該隊列是持久化的(durable = true)，一旦連接關閉或者客戶端退出，該排他隊列都會被自動刪除，這種隊列適用于一個客戶端同時發送和讀取消息的應用場景。
4. autoDelete： 是否自動刪除 ；如果autoDelete = true，當所有消費者都與這個隊列斷開連接時，這個隊列會自動刪除。注意： 不是說該隊列沒有消費者連接時該隊列就會自動刪除，因為當生產者聲明了該隊列且沒有消費者連接消費時，該隊列是不會自動刪除的。

5. arguments： 設置隊列的其他一些參數，如 x-rnessage-ttl 、x-expires 、x-rnax-length 、x-rnax-length-bytes、 x-dead-letter-exchange、 x-deadletter-routing-key 、 x-rnax-priority 。

參考：https://blog.csdn.net/AwayFuture/article/details/103405335?depth_1-utm_source=distribute.pc_relevant.none-task&utm_source=distribute.pc_relevant.none-task

下一篇https://blog.csdn.net/lilongwangyamin/article/details/105098011?rabbitmq 隊列

整體代碼：https://download.csdn.net/download/lilongwangyamin/12272687

總結

以上是生活随笔為你收集整理的RabbitMQ基础介绍的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。