在RabbitMQ中實(shí)現(xiàn)Work queues工作隊(duì)列模式

更新時(shí)間：2021年04月16日 15:00:37 作者：Java_Caiyo

這篇文章主要介紹了如何在RabbitMQ中實(shí)現(xiàn)Work queues模式,代碼詳細(xì),解釋清晰,可以幫助大家更好理解java,對(duì)這方面感興趣的朋友可以參考下

一、模式說(shuō)明

Work Queues 與入門(mén)程序的簡(jiǎn)單模式相比，多了一個(gè)或一些消費(fèi)端，多個(gè)消費(fèi)端共同消費(fèi)同一個(gè)隊(duì)列中的消息。

應(yīng)用場(chǎng)景 ：對(duì)于任務(wù)過(guò)重或任務(wù)較多情況使用工作隊(duì)列可以提高任務(wù)處理的速度。

二、代碼

Work Queues 與入門(mén)程序的簡(jiǎn)單模式的代碼是幾乎一樣的：可以完全復(fù)制，并復(fù)制多一個(gè)消費(fèi)者進(jìn)行多個(gè)消費(fèi)者同時(shí)消費(fèi)消息的測(cè)試。

①生產(chǎn)者

package com.itheima.rabbitmq.work; 
import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.Channel; 
import com.rabbitmq.client.Connection; 
import com.rabbitmq.client.ConnectionFactory; 
public class Producer { 
	static final String QUEUE_NAME = "work_queue"; 
	public static void main(String[] args) throws Exception { 
		//創(chuàng)建連接 
		Connection connection = ConnectionUtil.getConnection(); 
		// 創(chuàng)建頻道 
		Channel channel = connection.createChannel(); 
		// 聲明（創(chuàng)建）隊(duì)列 
		/**
		 * 參數(shù)1：隊(duì)列名稱(chēng) 
		 * 參數(shù)2：是否定義持久化隊(duì)列 
		 * 參數(shù)3：是否獨(dú)占本次連接 
		 * 參數(shù)4：是否在不使用的時(shí)候自動(dòng)刪除隊(duì)列 
		 * 參數(shù)5：隊(duì)列其它參數(shù) 
		*/ 
		channel.queueDeclare(QUEUE_NAME, true, false, false, null); 
		for (int i = 1; i <= 30; i++) { 
			// 發(fā)送信息 
			String message = "你好；小兔子！work模式--" + i; 
			/**
			 * 參數(shù)1：交換機(jī)名稱(chēng)，如果沒(méi)有指定則使用默認(rèn)Default Exchage 
			 * 參數(shù)2：路由key,簡(jiǎn)單模式可以傳遞隊(duì)列名稱(chēng) 
			 * 參數(shù)3：消息其它屬性 
			 * 參數(shù)4：消息內(nèi)容 
			*/ 
			channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); 
			System.out.println("已發(fā)送消息：" + message); 
		}
		// 關(guān)閉資源 
		channel.close(); connection.close(); 
	} 
}

②消費(fèi)者1

package com.itheima.rabbitmq.work; 
import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.*;
import java.io.IOException; 
public class Consumer1 { 
	public static void main(String[] args) throws Exception { 
		Connection connection = ConnectionUtil.getConnection(); 
		// 創(chuàng)建頻道 
		Channel channel = connection.createChannel(); 
		// 聲明（創(chuàng)建）隊(duì)列 
		/**
		 * 參數(shù)1：隊(duì)列名稱(chēng) 
		 * 參數(shù)2：是否定義持久化隊(duì)列 
		 * 參數(shù)3：是否獨(dú)占本次連接 
		 * 參數(shù)4：是否在不使用的時(shí)候自動(dòng)刪除隊(duì)列 
		 * 參數(shù)5：隊(duì)列其它參數(shù) 
		*/ 
		channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null); 
		//一次只能接收并處理一個(gè)消息 
		channel.basicQos(1); 
		//創(chuàng)建消費(fèi)者；并設(shè)置消息處理 
		DefaultConsumer consumer = new DefaultConsumer(channel){ 
			@Override 
			/**
			 * consumerTag 消息者標(biāo)簽，在channel.basicConsume時(shí)候可以指定 
			 * envelope 消息包的內(nèi)容，可從中獲取消息id，消息routingkey，交換機(jī)，消息和重傳標(biāo)志(收到消息失敗后是否需要重新發(fā)送) 
			 * properties 屬性信息 
			 * body 消息 
			*/ 
			public void handleDelivery(String consumerTag, Envelope envelope, 
					AMQP.BasicProperties properties, byte[] body) throws IOException { 
				try {
					//路由key 
					System.out.println("路由key為：" + envelope.getRoutingKey()); 
					//交換機(jī) 
					System.out.println("交換機(jī)為：" + envelope.getExchange()); 
					//消息id 
					System.out.println("消息id為：" + envelope.getDeliveryTag()); 
					//收到的消息 
					System.out.println("消費(fèi)者1-接收到的消息為：" + new String(body, "utf-8")); 
					Thread.sleep(1000); 
					//確認(rèn)消息 
					channel.basicAck(envelope.getDeliveryTag(), false); 
				} 
				catch (InterruptedException e) { 
					e.printStackTrace(); 
				} 
			} 
		};
		//監(jiān)聽(tīng)消息 
		/**
		 * 參數(shù)1：隊(duì)列名稱(chēng)
		 * 參數(shù)2：是否自動(dòng)確認(rèn)，設(shè)置為true為表示消息接收到自動(dòng)向mq回復(fù)接收到了，mq接收到回復(fù)會(huì)刪除消息，設(shè)置為false則需要手動(dòng)確認(rèn) 
		 * 參數(shù)3：消息接收到后回調(diào) 
		*/ 
		channel.basicConsume(Producer.QUEUE_NAME, false, consumer); 
	} 
}

③消費(fèi)者2

package com.itheima.rabbitmq.work; 
import com.itheima.rabbitmq.util.ConnectionUtil; 
import com.rabbitmq.client.*; 
import java.io.IOException; 
public class Consumer2 { 
	public static void main(String[] args) throws Exception { 
		Connection connection = ConnectionUtil.getConnection(); 
		// 創(chuàng)建頻道 
		Channel channel = connection.createChannel(); 
		// 聲明（創(chuàng)建）隊(duì)列 
		/**
		 * 參數(shù)1：隊(duì)列名稱(chēng) 
		 * 參數(shù)2：是否定義持久化隊(duì)列 
		 * 參數(shù)3：是否獨(dú)占本次連接 
		 * 參數(shù)4：是否在不使用的時(shí)候自動(dòng)刪除隊(duì)列 
		 * 參數(shù)5：隊(duì)列其它參數(shù) 
		*/ 
		channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null); 
		//一次只能接收并處理一個(gè)消息 
		channel.basicQos(1); 
		//創(chuàng)建消費(fèi)者；并設(shè)置消息處理 
		DefaultConsumer consumer = new DefaultConsumer(channel){ 
			@Override 
			/**
			 * consumerTag 消息者標(biāo)簽，在channel.basicConsume時(shí)候可以指定 
			 * envelope 消息包的內(nèi)容，可從中獲取消息id，消息routingkey，交換機(jī)，消息和重傳標(biāo)志(收到消息失敗后是否需要重新發(fā)送) 
			 * properties 屬性信息 
			 * body 消息 
			*/ 
			public void handleDelivery(String consumerTag, Envelope envelope, 
					AMQP.BasicProperties properties, byte[] body) throws IOException { 
				try {
					//路由key 
					System.out.println("路由key為：" + envelope.getRoutingKey()); 
					//交換機(jī) 
					System.out.println("交換機(jī)為：" + envelope.getExchange()); 
					//消息id 
					System.out.println("消息id為：" + envelope.getDeliveryTag());
					//收到的消息 
					System.out.println("消費(fèi)者2-接收到的消息為：" + new String(body, "utf-8")); 
					Thread.sleep(1000); 
					//確認(rèn)消息 
					channel.basicAck(envelope.getDeliveryTag(), false); 
				} catch (InterruptedException e) { 
					e.printStackTrace(); 
				} 
			} 
		};
		//監(jiān)聽(tīng)消息 
		/**
		 * 參數(shù)1：隊(duì)列名稱(chēng) 
		 * 參數(shù)2：是否自動(dòng)確認(rèn)，設(shè)置為true為表示消息接收到自動(dòng)向mq回復(fù)接收到了，mq接收到回復(fù)會(huì)刪除消息，設(shè)置為false則需要手動(dòng)確認(rèn) 
		 * 參數(shù)3：消息接收到后回調(diào) 
		*/ 
		channel.basicConsume(Producer.QUEUE_NAME, false, consumer); 
	} 
}

三、測(cè)試

啟動(dòng)兩個(gè)消費(fèi)者，然后再啟動(dòng)生產(chǎn)者發(fā)送消息；到IDEA的兩個(gè)消費(fèi)者對(duì)應(yīng)的控制臺(tái)查看是否競(jìng)爭(zhēng)性的接收到消息。

總結(jié)

在一個(gè)隊(duì)列中如果有多個(gè)消費(fèi)者，那么消費(fèi)者之間對(duì)于同一個(gè)消息的關(guān)系是競(jìng)爭(zhēng)的關(guān)系。

到此這篇關(guān)于如何在RabbitMQ中實(shí)現(xiàn)Work queues模式的文章就介紹到這了,希望對(duì)你有所幫助，更多相關(guān)RabbitMQ內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持腳本之家！

您可能感興趣的文章: