文章目錄

• 1_RabbitMQ初入門
• • 1.RabbitMQ的介紹
  • • 1.工作原理&發送/接收消息的流程
    • 2. 為什么要使用消息隊列？
    • 3.使用了消息隊列有什么缺點？
  • 2.安裝RabbitMQ
  • 3.入門程序Hello_消費者&生產者_
  • • 1.導入依賴
    • 2.生產者
    • • 1.設置連接信息
      • 2.獲取connection（連接），channel（信道）
      • 3.channel（信道）綁定隊列&實現消費方法
      • 4.接收消息
      • 完整代碼：
    • 3.消費者
    • • 1.設置連接信息
      • 2.獲取connection（連接），channel（信道）
      • 3.channel（信道）綁定隊列
      • 4.發送消息
      • 完整代碼：

1_RabbitMQ初入門

1.RabbitMQ的介紹

1.工作原理&發送/接收消息的流程

• Broker：消息隊列服務進程，此進程包括兩個部分：Exchange和Queue。

• Exchange：消息隊列交換機，按一定的規則將消息路由轉發到某個隊列，對消息進行過慮。

• Queue：消息隊列，存儲消息的隊列，消息到達隊列并轉發給指定的消費方。

• Producer：消息生產者，即生產方客戶端，生產方客戶端將消息發送到MQ。

• Consumer：消息消費者，即消費方客戶端，接收MQ轉發的消息。

  消息發布接收流程：

-----發送消息----- 1、生產者和Broker建立TCP連接。 2、生產者和Broker建立通道。 3、生產者通過通道消息發送給Broker，由Exchange將消息進行轉發。 4、Exchange將消息轉發到指定的Queue（隊列）----接收消息----- 1、消費者和Broker建立TCP連接 2、消費者和Broker建立通道 3、消費者監聽指定的Queue（隊列） 4、當有消息到達Queue時Broker默認將消息推送給消費者。 5、消費者接收到消息。

2. 為什么要使用消息隊列？

解耦，異步，削峰

3.使用了消息隊列有什么缺點？

分析：一個使用了MQ的項目，如果連這個問題都沒有考慮過，就把MQ引進去了，那就給自己的項目帶來了風險。我們引入一個技術，要對這個技術的弊端有充分的認識，才能做好防御。

系統的可用性降低：如果消息隊列掛了，那么系統也會受到影響

系統的復雜性增加：要多考慮很多方面的問題，比如一致性問題、如何保證消息不被重復消費，如何保證消息可靠傳輸。因此，需要考慮的東更多，系統的復雜性增大。

2.安裝RabbitMQ

詳情請參考此鏈接

3.入門程序Hello_消費者&生產者_

1.導入依賴

<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> </dependency><!-- 這個是日志包的依賴 --> <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-logging</artifactId> </dependency>

2.生產者

1.設置連接信息

通過ConnectionFactory獲取連接信息

//通過連接工廠創建新的連接和mq建立連接 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("127.0.0.1"); connectionFactory.setPort(5672);//端口 connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); //設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mq connectionFactory.setVirtualHost("/");

2.獲取connection（連接），channel（信道）

Connection connection = null; Channel channel = null; //建立新連接 connection = connectionFactory.newConnection(); //創建會話通道,生產者和mq服務所有通信都在channel通道中完成 channel = connection.createChannel();

3.channel（信道）綁定隊列&實現消費方法

//隊列 private static final String QUEUE = "helloworld";//監聽隊列 //聲明隊列，如果隊列在mq 中沒有則要創建 //參數：String queue, boolean durable, boolean exclusive, boolean autoDelete, M /*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/ channel.queueDeclare(QUEUE,true,false,false,null);//實現消費方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 當接收到消息后此方法將被調用* @param consumerTag 消費者標簽，用來標識消費者的，在監聽隊列時設置channel.basicConsume* @param envelope 信封，通過envelope* @param properties 消息屬性* @param body 消息內容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.B//交換機String exchange = envelope.getExchange();//消息id，mq在channel中用來標識消息的id，可用于確認消息已接收long deliveryTag = envelope.getDeliveryTag();//消息內容String message= new String(body,"utf-8");System.out.println("receive message:"+message);} };

4.接收消息

//監聽隊列//參數：String queue, boolean autoAck, Consumer callback/*** 參數明細：* 1、queue 隊列名稱* 2、autoAck 自動回復，當消費者接收到消息后要告訴mq消息已接收，如果將此參數設置為tru表示會自動回復mq，如果設置為false要通過編程實現回復* 3、callback，消費方法，當消費者接收到消息要執行的方法*/channel.basicConsume(QUEUE,true,defaultConsumer);

結果截圖：

完整代碼：

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Consumer01 {//隊列private static final String QUEUE = "helloworld";public static void main(String[] args) throws IOException, TimeoutException {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");//建立新連接Connection connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成Channel channel = connection.createChannel();//監聽隊列//聲明隊列，如果隊列在mq 中沒有則要創建//參數：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE,true,false,false,null);//實現消費方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 當接收到消息后此方法將被調用* @param consumerTag 消費者標簽，用來標識消費者的，在監聽隊列時設置channel.basicConsume* @param envelope 信封，通過envelope* @param properties 消息屬性* @param body 消息內容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交換機String exchange = envelope.getExchange();//消息id，mq在channel中用來標識消息的id，可用于確認消息已接收long deliveryTag = envelope.getDeliveryTag();//消息內容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//監聽隊列//參數：String queue, boolean autoAck, Consumer callback/*** 參數明細：* 1、queue 隊列名稱* 2、autoAck 自動回復，當消費者接收到消息后要告訴mq消息已接收，如果將此參數設置為tru表示會自動回復mq，如果設置為false要通過編程實現回復* 3、callback，消費方法，當消費者接收到消息要執行的方法*/channel.basicConsume(QUEUE,true,defaultConsumer);} }

3.消費者

1.設置連接信息

通過ConnectionFactory獲取連接信息

//通過連接工廠創建新的連接和mq建立連接 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("127.0.0.1"); connectionFactory.setPort(5672);//端口 connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); //設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mq connectionFactory.setVirtualHost("/");

2.獲取connection（連接），channel（信道）

Connection connection = null; Channel channel = null; //建立新連接 connection = connectionFactory.newConnection(); //創建會話通道,生產者和mq服務所有通信都在channel通道中完成 channel = connection.createChannel();

3.channel（信道）綁定隊列

//隊列 private static final String QUEUE = "helloworld";//聲明隊列，如果隊列在mq 中沒有則要創建 //參數：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments /*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/ channel.queueDeclare(QUEUE,true,false,false,null);

4.發送消息

//參數：String exchange, String routingKey, BasicProperties props, byte[] bod /*** 參數明細：* 1、exchange，交換機，如果不指定將使用mq的默認交換機（設置為""）* 2、routingKey，路由key，交換機根據路由key來將消息轉發到指定的隊列，如果使用默認交換機，routingKey設置為隊列的名稱* 3、props，消息的屬性* 4、body，消息內容*/ //消息內容 String message = "hello world 黑馬程序員"; channel.basicPublish("",QUEUE,null,message.getBytes()); System.out.println("send to mq "+message);

結果截圖：

完整代碼：

package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Producer01 {//隊列private static final String QUEUE = "helloworld";public static void main(String[] args) {//通過連接工廠創建新的連接和mq建立連接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//設置虛擬機，一個mq服務可以設置多個虛擬機，每個虛擬機就相當于一個獨立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新連接connection = connectionFactory.newConnection();//創建會話通道,生產者和mq服務所有通信都在channel通道中完成channel = connection.createChannel();//聲明隊列，如果隊列在mq 中沒有則要創建//參數：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 參數明細* 1、queue 隊列名稱* 2、durable 是否持久化，如果持久化，mq重啟后隊列還在* 3、exclusive 是否獨占連接，隊列只允許在該連接中訪問，如果connection連接關閉隊列則自動刪除,如果將此參數設置true可用于臨時隊列的創建* 4、autoDelete 自動刪除，隊列不再使用時是否自動刪除此隊列，如果將此參數和exclusive參數設置為true就可以實現臨時隊列（隊列不用了就自動刪除）* 5、arguments 參數，可以設置一個隊列的擴展參數，比如：可設置存活時間*/channel.queueDeclare(QUEUE,true,false,false,null);//發送消息//參數：String exchange, String routingKey, BasicProperties props, byte[] body/*** 參數明細：* 1、exchange，交換機，如果不指定將使用mq的默認交換機（設置為""）* 2、routingKey，路由key，交換機根據路由key來將消息轉發到指定的隊列，如果使用默認交換機，routingKey設置為隊列的名稱* 3、props，消息的屬性* 4、body，消息內容*///消息內容String message = "hello world 黑馬程序員";channel.basicPublish("",QUEUE,null,message.getBytes());System.out.println("send to mq "+message);} catch (Exception e) {e.printStackTrace();} finally {//關閉連接//先關閉通道try {channel.close();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}try {connection.close();} catch (IOException e) {e.printStackTrace();}}} }

總結

以上是生活随笔為你收集整理的1_RabbitMQ初入门入门Hello消费者+生产者的全部內容，希望文章能夠幫你解決所遇到的問題。

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