2. 程式人生 > >Rabbitmq的五種模式和案例

Rabbitmq的五種模式和案例

阿新 • • 發佈：2018-12-25

1.simple模式

 

訊息生產者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);
		}
	}
}

相關推薦

Rabbitmq的五種模式和案例

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

RabbitMQ五種模式

佇列模式 五種佇列 Queue——簡單佇列 Work queue——Work模式 Publish/Subscribe——釋出/訂閱者模式 Routing——路由模式 Topics——

cpufreq 五種模式和手動提升效能指令碼

cpufreq是一個動態調整cpu頻率的模組，系統啟動時生成一個資料夾/sys/devices/system/cpu/cpu0/cpufreq/，裡面有幾個檔案，其中： scaling_min_freq代表最低頻率， scaling_max_freq代表最高頻率， scali

RabbitMQ五種訊息佇列學習（五）--路由模式

RabbitMQ五種訊息佇列學習（五）–路由模式 標籤（空格分隔）： RabbitMQ 釋出訂閱模式只是利用路由這個功能，將訊息全部分發給所有與路由繫結的佇列中，不能對不同的訊息進行選擇性的分發。 模型結構 路由根據Routing Key進行

web優化必須瞭解的原理之I/o的五種模型和web的三種工作模式

=========================================== 圖解五種I/O模型 圖解web支援的三種工作模式 ===========================================  五種I/O：  1）阻塞I

python3開發進階-Django框架的中間件的五種用法和邏輯過程

meta alc 多個 img eth ble 方法 get action 閱讀目錄 什麽是中間件 中間件的執行流程 中間件的邏輯過程 一、什麽是中間件？ 官方的說法：中間件是一個用來處理Django的請求和響應的框架級別的鉤子。它是一個輕量、

Java進階篇設計模式之五-----外觀模式和裝飾器模式

和我 logs 適配器模式 del xtra implement () 實例化 網絡遊戲 前言 在上一篇中我們學習了結構型模式的適配器模式和橋接模式。本篇則來學習下結構型模式的外觀模式和裝飾器模式。 外觀模式 簡介 外觀模式隱藏系統的復雜性，並向客戶端提供了一個客戶端可以

字串中判斷存在的幾種模式和效率(string.contains、string.IndexOf、Regex.Match)

　通常情況下，我們判斷一個字串中是否存在某值常常會用string.contains，其實判斷一個字串中存在某值的方法有很多種，最常用的就是前述所說的string.contains，相對來說比較常用的還有string.IndexOf和Regex.Match。直接上程式碼，後面在說些什麼吧，通常情況下功能的實現最

訊息佇列五種模式介紹

簡單模式Hello World 功能：一個生產者P傳送訊息到佇列Q,一個消費者C接收 工作佇列模式Work Queue 功能：一個生產者，多個消費者，每個消費者獲取到的訊息唯一，多個消費者只有一個佇列 任務佇列：避免立即做一個資源密集型任務，必須等待它完成，而

mysql的五種日期和時間型別【轉載】

[mysql的五種日期和時間型別] mysql(5.5)所支援的日期時間型別有：DATETIME、 TIMESTAMP、DATE、TIME、YEAR。 幾種型別比較如下： 日期時間型別 佔用空間 日期格式 最小值 最大值 零值表示

mysql的五種日期和時間類型【轉載】

超出 14. 範圍 包含 最大 但是 header 影響 日期和時間類型 [mysql的五種日期和時間類型] mysql(5.5)所支持的日期時間類型有：DATETIME、 TIMESTAMP、DATE、TIME、YEAR。 幾種類型比較如下： 日期時間類型 占用空間

Linux中Vim編輯器三種模式和命令

原文地址：http://www.tuicool.com/articles/NZz2ii Vim的3種模式：一般模式、編輯模式和命令模式。 當我們是用“ vim filename ”編輯一個檔案的時候，預設進入該檔案的一般模式。 開啟檔案的命令有很多，如下圖所示。

白盒測試---邏輯路徑覆蓋的五種方法和物理路徑覆蓋(一)

一、語句路徑覆蓋：是一個比較弱的邏輯路徑覆蓋標準。是指通過選擇足夠的測試用例，使得執行這些用例時，被測程式的每一個語句至少被執行一次。 舉例：  測試用例  輸入 預期輸出  被測路徑

cpufreq 五種模式

cpufreq是一個動態調整cpu頻率的模組，系統啟動時生成一個資料夾/sys/devices/system/cpu/cpu0/cpufreq/，裡面有幾個檔案， 其中scaling_min_freq代表最低頻率，scaling_max_freq代表最高頻率，scali

盤點：U盤損壞後如何修復的五種狀況和解決方法

在這個數碼資訊時代，除了電腦、手機作為資訊傳遞的載體，還有一種行動式載體——U盤。相信不管是學生黨還是工作黨，都有使用過U盤進行拷貝檔案，使用過的小夥伴們就一定會遇到以下的幾種情況：U盤中毒、U盤檔案無法複製、U盤無顯示等5種，今天小編就來分享當遇到這些情況時，我們該如何解

詳解FSMO的五種角色和FSMO角色辨別方法及角色轉換方法

FSMO中文翻譯成操作主控，在說明FSMO的作用以前，先給大家介紹兩個概念： 單主複製：所謂的單主複製就是指從一個地方向其它地方進行復制，這個主要是用於以前的NT4域，我們知道，在NT4域的年代，域網路上區分PDC和BDC，所有的複製都是從PDC到BDC上進行的，因為NT4域用的是這種複製機構

Android任務棧的四種模式和應用場景

通過Acitivty的xml標籤來改變任務棧的預設行為 在清單檔案中使用android:launchMode="standard|singleInstance|singleTask|singleT

RabbitMQ的五種工作模式的程式碼實現

首先是一個SpringBoot專案，在專案中新增如下依賴：        1、簡單模式 package com.jt.test.rabbitmq; import org.junit.Before; import org.junit.Test;

設計模式2——單例模式（singleton），五種構建方法（最喜歡列舉和內部類方式）

單例模式，在配置檔案的時候用的比較多，各處的配置都保持統一性。 一、什麼是單例模式? layout title folder permalink categories tags

訊息佇列rabbitmq的五種工作模式（go語言版本）

前言：如果你對rabbitmq基本概念都不懂，可以移步此篇博文查閱訊息佇列RabbitMQ 一、單發單收 二、工作佇列Work Queue 三、釋出/訂閱 Publish/Subscribe 四、路由Routing 五、Topic型別的exchange 六、rabbitmq部分封裝程式碼及裝備工作