rabbitMQ模式

来源:https://www.cnblogs.com/matd/archive/2019/03/02/10463143.html
-Advertisement-
Play Games

1.hello 1.hello 1.hello 1.hello 消息生產者p將消息放入隊列 消費者監聽隊列,如果隊列中有消息,就消費掉,消息被拿走後,自動從隊列刪除(隱患,消息可能沒有被消費者正確處理,已經消失了,無法恢復) 應用場景:聊天室 案例： 1>.首先準備依賴 <dependency> < ...


1.hello

 消息生產者p將消息放入隊列

消費者監聽隊列,如果隊列中有消息,就消費掉,消息被拿走後,自動從隊列刪除
(隱患,消息可能沒有被消費者正確處理,已經消失了,無法恢復)

應用場景:聊天室 

案例：

1>.首先準備依賴

<dependency>  
     <groupId>org.springframework.boot</groupId>  
     <artifactId>spring-boot-starter-amqp</artifactId>  
</dependency>

 

2>.寫一個test類

 

public class SimpleTest {  
   //模擬生產者將消息放入隊列  
   @Test  
   public void send() throws Exception{  
       /*1 創建連接工廠 
        * 2 配置共創config 
        * 3 獲取連接 
        * 4獲取通道 
        * 5 從通道聲明queue 
        * 6 發送消息 
        * 7 釋放資源 
        */  
       ConnectionFactory factory=new ConnectionFactory();  
       factory.setHost("106.23.34.56");  
       factory.setPort(5672);  
       factory.setVirtualHost("/tb");  
       factory.setUsername("admin");  
       factory.setPassword("123456");  
       //從工廠獲取連接  
       Connection conn=factory.newConnection();  
       //從連接獲取通道  
       Channel chan=conn.createChannel();  
       //利用channel聲明第一個隊列  
       chan.queueDeclare("simple", false, false, false, null);  
       //queue String類型,表示聲明的queue對列的名字  
       //durable Boolean類型,表示是否持久化  
       //exclusive Boolean類型:當前聲明的queue是否專註;true當前連接創建的  
       //任何channle都可以連接這個queue,false,新的channel不可使用  
       //autoDelete Boolean類型:在最後連接使用完成後,是否刪除隊列,false  
       //arguments Map類型,其他聲明參數  
       //發送消息  
       String msg="helloworld,nihaoa";  
       chan.basicPublish("", "simple", null, msg.getBytes());  
       //exchange String類型,交換機名稱,簡單模式使用預設交換""  
       //routingkey String類型,當前的消息綁定的routingkey,簡單模式下,與隊列同名即可  
       //props BasicProperties類型,消息的屬性欄位對象,例如BasicProperties  
       //可以設置一個deliveryMode的值0 持久化,1 表示不持久化,durable配合使用  
       //body byte[] :消息字元串的byte數組  
   }  
   //模擬消費端  
   @Test  
   public void receive() throws Exception{

 

ConnectionFactory factory=new ConnectionFactory();  
factory.setHost("106.23.34.56");  
factory.setPort(5672);  
factory.setVirtualHost("/tb");  
factory.setUsername("admin");  
factory.setPassword("123456");  
//從工廠獲取連接

Connection conn=factory.newConnection();//從連接獲取通道Channel chan=conn.createChannel();chan.queueDeclare("simple", false, false, false, null);//創建一個消費者QueueingConsumer consumer= new QueueingConsumer(chan);chan.basicConsume("simple", consumer);//監聽隊列while(true){//獲取下一個delivery,delivery從隊列獲取消息Delivery delivery = consumer.nextDelivery();String msg=new String(delivery.getBody());System.out.println(msg);}}}

2.work模式

生產者將消息放入隊列
多個消費者同時監聽同一個隊列,消息如何被消費?
C1,C2共同爭搶當前消息隊列的內容,誰先拿到消息,誰來負責消費
應用場景:紅包;大型項目中的資源調度過程(直接由最空閑的系統爭搶到資源處理任務) 

案例：

1>首先寫一個工具類

public class ConnectionUtil {
 
 public static Connection getConn(){
   try{
     ConnectionFactory factory=new ConnectionFactory();
     factory.setHost("106.33.44.179");
     factory.setPort(5672);
     factory.setVirtualHost("/tb");
     factory.setUsername("admin");
     factory.setPassword("123456");
   
     //從工廠獲取連接
     Connection conn=factory.newConnection();
     return conn;
   }catch(Exception e){
     System.out.println(e.getMessage());
     return null;
   }
   
 }
}

 

2>寫test類

public class WorkTest {
 @Test
 public void send() throws Exception{
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //聲明隊列
   chan.queueDeclare("work", false, false, false, null);
   for(int i=0;i<100;i++){
     String msg="1712,hello:"+i+"message";
     chan.basicPublish("", "work", null, msg.getBytes());
     System.out.println("第"+i+"條信息已經發送");
   }
   chan.close();
   conn.close();
 }
 @Test
 public void receive1() throws Exception{
   //獲取連接,獲取通道
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   chan.queueDeclare("work", false, false, false, null);
   //同一時刻伺服器只發送一條消息給同一消費者,消費者空閑,才發送一條
   chan.basicQos(1);
   //定義消費者
   QueueingConsumer consumer=new QueueingConsumer(chan);
   //綁定隊列和消費者的關係
   //queue
   //autoAck:消息被消費後,是否自動確認回執,如果false,不自動需要手動在
   //完成消息消費後進行回執確認,channel.ack,channel.nack
   //callback
   //chan.basicConsume(queue, autoAck, callback)
   chan.basicConsume("work", false, consumer);
   //監聽
   while(true){
     Delivery delivery=consumer.nextDelivery();
     byte[] result = delivery.getBody();
     String msg=new String(result);
     System.out.println("接受到:"+msg);
     Thread.sleep(50);
     //返回伺服器,回執
     chan.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
   }  
 }
 @Test
 public void receive2() throws Exception{
   //獲取連接,獲取通道
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   chan.queueDeclare("work", false, false, false, null);
   //同一時刻伺服器只發送一條消息給同一消費者,消費者空閑,才發送一條
   chan.basicQos(1);
   //定義消費者
   QueueingConsumer consumer=new QueueingConsumer(chan);
   //綁定隊列和消費者的關係
   //queue
   //autoAck:消息被消費後,是否自動確認回執,如果false,不自動需要手動在
   //完成消息消費後進行回執確認,channel.ack,channel.nack
   //callback
   //chan.basicConsume(queue, autoAck, callback)
   chan.basicConsume("work", false, consumer);
   //監聽
   while(true){
     Delivery delivery=consumer.nextDelivery();
     byte[] result = delivery.getBody();
     String msg=new String(result);
     System.out.println("接受到:"+msg);
     Thread.sleep(150);
     //返回伺服器,回執
     chan.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
   }
 }
 
}

3 publish/fanout發佈訂閱


生產者將消息交給交換機
有交換機根據發佈訂閱的模式設定將消息同步到所有的綁定隊列中;
後端的消費者都能拿到消息

應用場景:郵件群發,群聊天,廣告

案例：

public class FanoutTest {
 //交換機,有類型,發佈訂閱:fanout
 //路由模式:direct
 //主題模式:topic
 @Test
 public void send() throws Exception {
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //聲明交換機
   //參數意義,1 交換機名稱,2 類型:fanout,direct,topic
   chan.exchangeDeclare("fanoutEx", "fanout");
   //發送消息
   for(int i=0;i<100;i++){
     String msg="1712 hello:"+i+"msg";
     chan.basicPublish("fanoutEx", "", null, msg.getBytes());
     System.out.println("第"+i+"條信息已經發送");
   }
 }
 
 @Test
 public void receiv01() throws Exception{
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //生命隊列
   chan.queueDeclare("fanout01", false, false, false, null);
   //聲明交換機
   chan.exchangeDeclare("fanoutEx", "fanout");
   //綁定隊列到交換機
   //參數 1 隊列名稱,2 交換機名稱 3 路由key
   chan.queueBind("fanout01", "fanoutEx", "");
   chan.basicQos(1);
   //定義消費者
   QueueingConsumer consumer=new QueueingConsumer(chan);
   //消費者與隊列綁定
   chan.basicConsume("fanout01",false, consumer);
   while(true){
     Delivery delivery= consumer.nextDelivery();
     System.out.println("一號消費者接收到"+
     new String(delivery.getBody()));
     chan.basicAck(delivery.getEnvelope().
         getDeliveryTag(), false);
   }
 }
 @Test
 public void receiv02() throws Exception{
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //生命隊列
   chan.queueDeclare("fanout02", false, false, false, null);
   //聲明交換機
   chan.exchangeDeclare("fanoutEx", "fanout");
   //綁定隊列到交換機
   //參數 1 隊列名稱,2 交換機名稱 3 路由key
   chan.queueBind("fanout02", "fanoutEx", "");
   chan.basicQos(1);
   //定義消費者
   QueueingConsumer consumer=new QueueingConsumer(chan);
   //消費者與隊列綁定
   chan.basicConsume("fanout02",false, consumer);
   while(true){
     Delivery delivery= consumer.nextDelivery();
     System.out.println("二號消費者接收到"+new String(delivery.getBody()));
     chan.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
   }
 }
}

4 routing路由模式

 

生產者發送消息到交換機,同時綁定一個路由Key,交換機根據路由key對下游綁定的隊列進行路
由key的判斷,滿足路由key的隊列才會接收到消息,消費者消費消息

應用場景: 項目中的error報錯

案例：

public class RoutingTopicTest {
 
 @Test
 public void routingSend() throws Exception{
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //聲明交換機
   //參數意義,1 交換機名稱,2 類型:fanout,direct,topic
   chan.exchangeDeclare("directEx", "direct");
   //發送消息
   String msg="路由模式的消息";
   chan.basicPublish("directEx", "jt1713", 
       null, msg.getBytes());
 }
 @Test
 public void routingRec01() throws Exception{
   System.out.println("一號消費者等待接收消息");
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //聲明隊列
   chan.queueDeclare("direct01", false, false, false, null);
   //聲明交換機
   chan.exchangeDeclare("directEx", "direct");
   //綁定隊列到交換機
   //參數 1 隊列名稱,2 交換機名稱 3 路由key
   chan.queueBind("direct01", "directEx", "jt1712");
   chan.basicQos(1);
   //定義消費者
   QueueingConsumer consumer=new QueueingConsumer(chan);
   //消費者與隊列綁定
   chan.basicConsume("direct01",false, consumer);
   while(true){
     Delivery delivery= consumer.nextDelivery();
     System.out.println("一號消費者接收到"+
     new String(delivery.getBody()));
     chan.basicAck(delivery.getEnvelope().
         getDeliveryTag(), false);
   }
 }
 @Test
 public void routingRec02() throws Exception{
   System.out.println("二號消費者等待接收消息");
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //聲明隊列
   chan.queueDeclare("direct02", false, false, false, null);
   //聲明交換機
   chan.exchangeDeclare("directEx", "direct");
   //綁定隊列到交換機
   //參數 1 隊列名稱,2 交換機名稱 3 路由key
   chan.queueBind("direct02", "directEx", "jt1711");
   chan.basicQos(1);
   //定義消費者
   QueueingConsumer consumer=new QueueingConsumer(chan);
   //消費者與隊列綁定
   chan.basicConsume("direct02",false, consumer);
   while(true){
     Delivery delivery= consumer.nextDelivery();
     System.out.println("二號消費者接收到"+
     new String(delivery.getBody()));
     chan.basicAck(delivery.getEnvelope().
         getDeliveryTag(), false);
   }
 }
}

5 topic主題模式

*號代表單個詞語
#代表多個詞語

其他的內容與routing路由模式一致

案例：

public class RoutingTopicTest {
 
 
 @Test
 public void routingRec02() throws Exception{
   System.out.println("二號消費者等待接收消息");
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //聲明隊列
   chan.queueDeclare("direct02", false, false, false, null);
   //聲明交換機
   chan.exchangeDeclare("directEx", "direct");
   //綁定隊列到交換機
   //參數 1 隊列名稱,2 交換機名稱 3 路由key
   chan.queueBind("direct02", "directEx", "jt1711");
   chan.basicQos(1);
   //定義消費者
   QueueingConsumer consumer=new QueueingConsumer(chan);
   //消費者與隊列綁定
   chan.basicConsume("direct02",false, consumer);
   while(true){
     Delivery delivery= consumer.nextDelivery();
     System.out.println("二號消費者接收到"+
     new String(delivery.getBody()));
     chan.basicAck(delivery.getEnvelope().
         getDeliveryTag(), false);
   }
 }
 
 @Test
 public void topicSend() throws Exception{
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //聲明交換機
   //參數意義,1 交換機名稱,2 類型:fanout,direct,topic
   chan.exchangeDeclare("topicEx", "topic");
   //發送消息
   String msg="主題模式的消息";
   chan.basicPublish("topicEx", "jt1712.add.update", 
       null, msg.getBytes());
 }
 @Test
 public void topicRec01() throws Exception{
   System.out.println("一號消費者等待接收消息");
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //聲明隊列
   chan.queueDeclare("topic01", false, false, false, null);
   //聲明交換機
   chan.exchangeDeclare("topicEx", "topic");
   //綁定隊列到交換機
   //參數 1 隊列名稱,2 交換機名稱 3 路由key
   chan.queueBind("topic01", "topicEx", "jt1712");
   chan.basicQos(1);
   //定義消費者
   QueueingConsumer consumer=new QueueingConsumer(chan);
   //消費者與隊列綁定
   chan.basicConsume("topic01",false, consumer);
   while(true){
     Delivery delivery= consumer.nextDelivery();
     System.out.println("一號消費者接收到"+
     new String(delivery.getBody()));
     chan.basicAck(delivery.getEnvelope().
         getDeliveryTag(), false);
   }
 }
 @Test
 public void topicRec02() throws Exception{
   System.out.println("二號消費者等待接收消息");
   //獲取連接
   Connection conn = ConnectionUtil.getConn();
   Channel chan = conn.createChannel();
   //聲明隊列
   chan.queueDeclare("topic02", false, false, false, null);
   //聲明交換機
   chan.exchangeDeclare("topicEx", "topic");
   //綁定隊列到交換機
   //參數 1 隊列名稱,2 交換機名稱 3 路由key
   chan.queueBind("topic02", "topicEx", "jt1712.#");
   chan.basicQos(1);
   //定義消費者
   QueueingConsumer consumer=new QueueingConsumer(chan);
   //消費者與隊列綁定
   chan.basicConsume("topic02",false, consumer);
   while(true){
     Delivery delivery= consumer.nextDelivery();
     System.out.println("二號消費者接收到"+
     new String(delivery.getBody()));
     chan.basicAck(delivery.getEnvelope().
         getDeliveryTag(), false);
   }
 }
}

您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • iframe中的a標簽電話鏈接不能正常打開
    背景 經測試，android手機中沒有這個問題， iphone手機中的Safari瀏覽器會出現這個問題。 例如： 解決辦法： 給鏈接加上 target="_parent"， 如果iframe的嵌套比較深可以用 target="_top" ...
  • 迅速搭建簡易靜態伺服器
    有時直接打開本地html文件會使一些web操作無法進行，需要運行一個本地伺服器。 使用nodejs的 可以迅速地啟動一個本地靜態資源伺服器 ...
  • AntD框架的upload組件上傳圖片時遇到的一些坑
    前言 本次做後臺管理系統，採用的是 AntD 框架。涉及到圖片的上傳，用的是AntD的 "upload" 組件。 前端做文件上傳這個功能，是很有技術難度的。既然框架給我們提供好了，那就直接用唄。結果用的時候，發現 upload 組件的很多bug。下麵來列舉幾個。 備註：本文寫於2019 03 02， ...
  • 亞馬遜 amazon connect（呼叫中心）
    背景 公司為提高客服部門溝通效率對接電話呼叫中心，調研後選擇了亞馬遜的Amazon Connect服務，因為是國外業務沒有選擇用阿裡雲，怕有坑。 Amazon Connect後臺 需要在後臺創建“聯繫流”，也就是用戶接通電話後我們提供的一系列功能選項，比如開始放一段歡迎語音，然後提示用戶選擇1，2， ...
  • git命令詳解（一）
    今天我們來詳解一下git的各種命令,此為git的第一篇，後續還會有好幾篇，希望大家看了能有所進步 Git Commit Git 倉庫中的提交記錄保存的是你的目錄下所有文件的快照，就像是把整個目錄複製，然後再粘貼一樣，但比複製粘貼優雅許多！ Git 希望提交記錄儘可能地輕量，因此在你每次進行提交時，它 ...
  • 關於JavaScript閉包的粗淺理解
    在JavaScript中，使用var創建變數，會創建全局變數或局部變數。 只有在非函數內創建的變數，才是全局變數，該變數可以在任何地方被讀取。 而在函數內創建變數時，只有在函數內部才可讀取。在函數外部時，調用函數也無法讀取局部變數。 function test(){ var g = 5; } // ...
  • 計算屬性
    計算屬性 1.1 什麼是計算屬性: 插值表達式常用於簡單的運算,當其過長或邏輯複雜時,會難以維護,這時應該使用計算屬性。 插值表達式里的值是JS表達式 所有的計算屬性都以函數的形式寫在Vue實例內的computed選項內,最終返回計算後的結果。 1.2 計算屬性的用法 在一個計算屬性里可以完成各種復 ...
  • 項目分層-----MVC
    MVC設計模式：modle層，view層，controller層 以前學習的servlet其實就是一個java類，或者說經過規範的java類，實際進行跳轉時，還是要在web.xml文件中配置才能正常跳轉。 controller層可以放servlet，在SpringMVC中則可以創建java類通過＠c ...
一周排行
    -Advertisement-
    Play Games
  • WPF 已知問題 監聽 WMI 事件導致觸摸失效
    示例項目結構 在 Visual Studio 中創建一個 WinForms 應用程式後，項目結構如下所示： MyWinFormsApp/ │ ├───Properties/ │ └───Settings.settings │ ├───bin/ │ ├───Debug/ │ └───Release/ ...
  • [STAThread]特性
    [STAThread] 特性用於需要與 COM 組件交互的應用程式，尤其是依賴單線程模型（如 Windows Forms 應用程式）的組件。在 STA 模式下，線程擁有自己的消息迴圈，這對於處理用戶界面和某些 COM 組件是必要的。 [STAThread] static void Main(stri ...
  • UWP WinUI3 傳入 AddHandler 的 RoutedEventHandler 類型與事件所需不匹配將拋出參數異常
    在WinForm中使用全局異常捕獲處理 在WinForm應用程式中，全局異常捕獲是確保程式穩定性的關鍵。通過在Program類的Main方法中設置全局異常處理，可以有效地捕獲並處理未預見的異常，從而避免程式崩潰。 註冊全局異常事件 [STAThread] static void Main() { / ...
  • .NET 多版本 WinForm 開源控制項庫 SunnyUI
    前言 給大家推薦一款開源的 Winform 控制項庫，可以幫助我們開發更加美觀、漂亮的 WinForm 界面。 項目介紹 SunnyUI.NET 是一個基於 .NET Framework 4.0+、.NET 6、.NET 7 和 .NET 8 的 WinForm 開源控制項庫，同時也提供了工具類庫、擴展 ...
  • .net core8 使用Swagger（附當前源碼）
    說明 該文章是屬於OverallAuth2.0系列文章，每周更新一篇該系列文章（從0到1完成系統開發）。 該系統文章，我會儘量說的非常詳細，做到不管新手、老手都能看懂。 說明：OverallAuth2.0 是一個簡單、易懂、功能強大的許可權+可視化流程管理系統。 有興趣的朋友，請關註我吧(*^▽^*) ...
  • WPF 什麼時候 VisualTreeHelper.GetDescendantBounds 將返回無窮大
    一、下載安裝 1.下載git 必須先下載並安裝git，再TortoiseGit下載安裝 git安裝參考教程：https://blog.csdn.net/mukes/article/details/115693833 2.TortoiseGit下載與安裝 TortoiseGit，Git客戶端，32/6 ...
  • C# 開源教程帶你輕鬆掌握數據結構與演算法
    前言 在項目開發過程中，理解數據結構和演算法如同掌握蓋房子的秘訣。演算法不僅能幫助我們編寫高效、優質的代碼，還能解決項目中遇到的各種難題。 給大家推薦一個支持C#的開源免費、新手友好的數據結構與演算法入門教程：Hello演算法。 項目介紹 《Hello Algo》是一本開源免費、新手友好的數據結構與演算法入門 ...
  • 第22篇 生成proto文件bat腳本
    1.生成單個Proto.bat內容 @rem Copyright 2016, Google Inc. @rem All rights reserved. @rem @rem Redistribution and use in source and binary forms, with or with ...
  • 記一次 .NET某上位機視覺程式 卡死分析
    一：背景 1. 講故事 前段時間有位朋友找到我，說他的窗體程式在客戶這邊出現了卡死，讓我幫忙看下怎麼回事？dump也生成了，既然有dump了那就上 windbg 分析吧。 二：WinDbg 分析 1. 為什麼會卡死 窗體程式的卡死，入口門檻很低，後續往下分析就不一定了，不管怎麼說先用 !clrsta ...
  • 醫學DICOM文件與PACS系統進行數據傳輸
    前言 人工智慧時代，人臉識別技術已成為安全驗證、身份識別和用戶交互的關鍵工具。 給大家推薦一款.NET 開源提供了強大的人臉識別 API，工具不僅易於集成，還具備高效處理能力。 本文將介紹一款如何利用這些API，為我們的項目添加智能識別的亮點。 項目介紹 GitHub 上擁有 1.2k 星標的 C# ...
所有分類