消息中間件——RabbitMQ（八）高級特性全在這裏!（下）

時間  2019-11-16
標籤 消息 中間件 rabbitmq 高級 特性 這裏 欄目 RabbitMQ 简体版
原文   原文鏈接

求關注

高級特性全在這裏!（下）

前言

上一篇消息中間件——RabbitMQ（七）高級特性全在這裏!（上）中咱們介紹了消息如何保障100%的投遞成功？,冪等性概念詳解,在海量訂單產生的業務高峯期，如何避免消息的重複消費的問題？,Confirm確認消息、Return返回消息。這篇咱們來介紹下下面內容。java

  • 自定義消費者
  • 消息的限流（防止佔用內存過多，節點宕機）
  • 消息的ACK與重回隊列
  • TTL消息
  • 死信隊列

1. 自定義消費者

1.1 消費端自定義監聽

咱們通常就在代碼中編寫while循環，進行consumer.nextDelivery方法進行獲取下一條消息，而後進行消費處理！git

可是這種輪訓的方式確定是很差的，代碼也比較low。github

  • 咱們使用自定義的Consumer更加的方便，解耦性更加的強，也是在實際工做中最多見的使用方式！

消費端自定義監聽

1.2 代碼演示

1.2.1 生產者

/** * * @ClassName: Producer * @Description: 生產者 * @author Coder編程 * @date2019年7月30日 下午23:15:51 * */
public class Producer {

	
	public static void main(String[] args) throws Exception {
		
		//1 建立ConnectionFactory
		Connection connection = ConnectionUtils.getConnection();
		Channel channel = connection.createChannel();
		
		String exchange = "test_consumer_exchange";
		String routingKey = "consumer.save";
		
		String msg = "Hello RabbitMQ Consumer Message";
		
		for(int i =0; i<5; i ++){
			channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
		}
		
	}
}


複製代碼

1.2.2 消費者

/** * * @ClassName: Consumer * @Description: 消費者 * @author Coder編程 * @date2019年7月30日 下午23:13:51 * */
public class Consumer {

	
	public static void main(String[] args) throws Exception {
		
		
		// 建立ConnectionFactory
		Connection connection = ConnectionUtils.getConnection();
		Channel channel = connection.createChannel();
		
		
		String exchangeName = "test_consumer_exchange";
		String routingKey = "consumer.#";
		String queueName = "test_consumer_queue";
		
		channel.exchangeDeclare(exchangeName, "topic", true, false, null);
		channel.queueDeclare(queueName, true, false, false, null);
		channel.queueBind(queueName, exchangeName, routingKey);
		
		//實現本身的MyConsumer()
		channel.basicConsume(queueName, true, new MyConsumer(channel));
	}
}


複製代碼

1.2.3 自定義類：MyConsumer

/** * * @ClassName: MyConsumer * @Description: TODO * @author Coder編程 * @date 2019年7月30日 下午23:11:55 * */
public class MyConsumer extends DefaultConsumer {


	public MyConsumer(Channel channel) {
		super(channel);
	}

	//根據需求，重寫本身須要的方法。
	@Override
	public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
		System.err.println("-----------consume message----------");
		//消費標籤
		System.err.println("consumerTag: " + consumerTag);
		//這個對象包含許多關鍵信息
		System.err.println("envelope: " + envelope);
		System.err.println("properties: " + properties);
		System.err.println("body: " + new String(body));
	}


}


複製代碼

1.3 打印結果

打印結果

2. 消費端限流

2.1 什麼是消費端的限流？

  • 假設一個場景，首先，咱們Rabbitmq服務器有上萬條未處理的消息，咱們隨便打開一個消費者客戶端，會出現下面狀況：
  • 巨量的消息瞬間所有推送過來，可是咱們單個客戶端沒法同時處理這麼多數據！這個時候很容易致使服務器崩潰，出現故障。

爲何不在生產端進行限流呢？面試

由於在高併發的狀況下，客戶量就是很是大，因此很難在生產端作限制。所以咱們能夠用MQ在消費端作限流。編程

  • RabbitMQ提供了一種qos(服務質量保證)功能，即在非自動確認消息的前提下，若是必定數目的消息（經過基於consume或者channel設置Qos的值）未被確認前，不進行消費新的消息。 在限流的狀況下，千萬不要設置自動簽收，要設置爲手動簽收。
  • void BasicQos(uint prfetchSize,ushort prefetchCount,bool global);

參數解釋： prefetchSize:0 prefetchCount:會告訴RabbitMQ不要同時給一個消費者推送多於N個消息，即一旦有N個消息尚未ack，則該consumer將block掉，直到有消息ack。 global: true\false 是否將上面設置應用於channel，簡單點說，就是上面限制是channel級別仍是consumer級別。 prefetchSize和global這兩項，rabbitmq沒有實現，暫且不研究prefetch_count在no_ask = false的狀況下生效，即在自動應答的狀況下這兩個值是不生效的。bash

第一個參數：消息的限制大小，消息多少兆。通常不作限制，設置爲0 第二個參數：一次最多處理多少條，實際工做中設置爲1就好 第三個參數：限流策略在什麼上應用。在RabbitMQ通常有兩個應用級別：1.通道 2.Consumer級別。通常設置爲false，true 表示channel級別，false表示在consumer級別服務器

2.2 代碼演示

2.2.1 生產者

/** * * @ClassName: Producer * @Description: 生產者 * @author Coder編程 * @date2019年7月30日 下午23:15:51 * */
public class Producer {

	
	public static void main(String[] args) throws Exception {
		
		//1 建立ConnectionFactory
		Connection connection = ConnectionUtils.getConnection();
		Channel channel = connection.createChannel();
		
		String exchange = "test_qos_exchange";
		String routingKey = "qos.save";
		
		String msg = "Hello RabbitMQ QOS Message";
		
		for(int i =0; i<5; i ++){
			channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
		}
		
	}
}


複製代碼

2.2.2 消費者

/** * * @ClassName: Consumer * @Description: 消費者 * @author Coder編程 * @date2019年7月30日 下午23:13:51 * */
public class Consumer {

	
	public static void main(String[] args) throws Exception {
		
		
		//1 建立ConnectionFactory
		Connection connection = ConnectionUtils.getConnection();
		Channel channel = connection.createChannel();
		
		
		String exchangeName = "test_qos_exchange";
		String queueName = "test_qos_queue";
		String routingKey = "qos.#";
		
		channel.exchangeDeclare(exchangeName, "topic", true, false, null);
		channel.queueDeclare(queueName, true, false, false, null);
		channel.queueBind(queueName, exchangeName, routingKey);
		
		//1 限流方式 第一件事就是 autoAck設置爲 false
		//設置爲1，表示一條一條數據處理
		channel.basicQos(0, 1, false);
		
		channel.basicConsume(queueName, false, new MyConsumer(channel));
		
		
	}
}


複製代碼

2.2.3 自定義類：MyConsumer

/** * * @ClassName: MyConsumer * @Description: TODO * @author Coder編程 * @date 2019年7月30日 下午23:11:55 * */
public class MyConsumer extends DefaultConsumer {


	private Channel channel ;
	
	public MyConsumer(Channel channel) {
		super(channel);
		this.channel = channel;
	}

	@Override
	public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
		System.err.println("-----------consume message----------");
		System.err.println("consumerTag: " + consumerTag);
		System.err.println("envelope: " + envelope);
		System.err.println("properties: " + properties);
		System.err.println("body: " + new String(body));
		
		//須要作簽收,false表示不支持批量簽收
		channel.basicAck(envelope.getDeliveryTag(), false);
		
	}


}


複製代碼

2.2.4 測試結果

咱們先註釋掉：channel.basicAck(envelope.getDeliveryTag(), false);而後啓動Consumer。 查看Exchange微信

Exchange

查看Queues 併發

Queues

而後再啓動Producer。查看打印結果： ide

打印結果

咱們會發現消費端，只收到了一條消息。這是爲何呢？

第一點由於咱們在consumer中

channel.basicConsume(queueName, false, new MyConsumer(channel));
複製代碼

第二個參數設置爲false爲手動簽收。

第二點在qos中設置只接受一條消息。若是這一條消息不給Broker Ack應答的話，那麼Broker會認爲你並無消費完這一條消息，那麼就不會繼續發送消息。

channel.basicQos(0, 1, false);
複製代碼

能夠看下管控臺，unack=1，Ready=4，total=5.

queues

接下來咱們放開註釋channel.basicAck(envelope.getDeliveryTag(), false); 進行消息簽收。重啓服務。

3.1 打印結果

打印結果
能夠看到正常打印五條結果

4. 消費端ACK與重回隊列

4.1 消費端的手工ACK和NACK

消費端進行消費的時候，若是因爲業務異常咱們能夠進行日誌的記錄，而後進行補償！

若是因爲服務器宕機等嚴重問題，那咱們就須要手工進行ACK保障消費端消費成功！

4.2 消費端的重回隊列

消費端重回隊列是爲了對沒有處理成功的消息，把消息從新傳遞給Broker!

通常咱們在實際應用中，都會關閉重回隊列，也就是設置爲False.

4.3 代碼演示

4.3.1 生產者

/** * * @ClassName: Producer * @Description: 生產者 * @author Coder編程 * @date2019年7月30日 下午23:15:51 * */
public class Producer {

	
	public static void main(String[] args) throws Exception {
		
		//1建立ConnectionFactory
		Connection connection = ConnectionUtils.getConnection();
		Channel channel = connection.createChannel();
		
		String exchange = "test_ack_exchange";
		String routingKey = "ack.save";
		
		
		
		for(int i =0; i<5; i ++){
			
			Map<String, Object> headers = new HashMap<String, Object>();
			headers.put("num", i);
			
			//添加屬性，後續會使用到
			AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
					.deliveryMode(2) //投遞模式，持久化
					.contentEncoding("UTF-8")
					.headers(headers)
					.build();
			String msg = "Hello RabbitMQ ACK Message " + i;
			channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());
		}
		
	}
}


複製代碼

4.3.2 消費者

/** * * @ClassName: Consumer * @Description: 消費者 * @author Coder編程 * @date2019年7月30日 下午23:13:51 * */
public class Consumer {

	
	public static void main(String[] args) throws Exception {
		
		
		//1建立ConnectionFactory
		Connection connection = ConnectionUtils.getConnection();
		Channel channel = connection.createChannel();
		
		
		String exchangeName = "test_ack_exchange";
		String queueName = "test_ack_queue";
		String routingKey = "ack.#";
		
		channel.exchangeDeclare(exchangeName, "topic", true, false, null);
		channel.queueDeclare(queueName, true, false, false, null);
		channel.queueBind(queueName, exchangeName, routingKey);
		
		// 手工簽收 必需要關閉 autoAck = false
		channel.basicConsume(queueName, false, new MyConsumer(channel));
		
		
	}
}


複製代碼

4.3.3 自定義類：MyConsumer

/** * * @ClassName: MyConsumer * @Description: TODO * @author Coder編程 * @date 2019年7月30日 下午23:11:55 * */
public class MyConsumer extends DefaultConsumer {


	private Channel channel ;
	
	public MyConsumer(Channel channel) {
		super(channel);
		this.channel = channel;
	}

	@Override
	public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
		System.err.println("-----------consume message----------");
		System.err.println("body: " + new String(body));
		try {
			Thread.sleep(2000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		if((Integer)properties.getHeaders().get("num") == 0) {
			//Nack三個參數 第二個參數：是不是批量，第三個參數：是否重回隊列（須要注意可能會發生重複消費，形成死循環）
			channel.basicNack(envelope.getDeliveryTag(), false, true);
		} else {
			channel.basicAck(envelope.getDeliveryTag(), false);
		}
		
	}


}


複製代碼

5.1 打印結果：

打印結果

注意： 能夠看到重回隊列會出現重複消費致使死循環的問題，這時候最好設置重試次數，好比超過三次後，消息仍是消費失敗，就將消息丟棄。

6. TTL隊列/消息

6.1 TTL

  • TTL是Time To Live的縮寫，也就是生存時間
  • RabbitMQ支持消息的過時時間，在消息發送時能夠進行指定
  • RabbitMQ支持隊列的過時時間，從消息入隊列開始計算，只要超過了隊列的超時時間配置，那麼消息會自動的清除

6.2 代碼演示

6.2.1 直接經過管控臺進行演示

添加queue
經過管控臺建立一個隊列

x-max-length 隊列的最大大小 x-message-ttl 設置10秒鐘，若是消息尚未被消費的話，就會被清除。

queue添加成功
添加exchange
添加exchange
exchange添加成功
Queue與Exchange進行綁定

點擊 test_ttl_exchange 進行綁定

進行綁定
查看是否綁定成功

綁定1
綁定2
經過管控臺發送消息
發送消息
queue中有消息
消息未處理自動清除
十秒後消息消失

生產端設置過時時間

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
				.deliveryMode(2)
				.contentEncoding("UTF-8")
				.expiration("10000")
				.headers(headers)
				.build();

複製代碼

這兩個屬性並不相同，一個對應的是消息體，一個對應的是隊列的過時。

7. 死信隊列

7.1 概念理解

死信隊列：DLX,Dead-Letter-Exchange RabbitMQ的死信隊裏與Exchange息息相關

  • 利用DLX,當消息在一個隊列中變成死信（dead message）以後，它能被從新publish到另外一個Exchange，這個Exchange就是DLX

消息變成死信有如下幾種狀況

  • 消息被拒絕（basic.reject/basic.nack）而且requeue=false
  • 消息TTL過時
  • 隊列達到最大長度

DLX也是一個正常的Exchange，和通常的Exchange沒有區別，它能在任何的隊列上被指定，實際上就是設置某個隊列的屬性

當這個隊列中有死信時，RabbitMQ就會自動的將這個消息從新發布到設置的Exchange上去，進而被路由到另外一個隊列。

能夠監聽這個隊列中消息作相應的處理，這個特徵能夠彌補RabbitMQ3.0之前支持的immediate參數的功能。

7.2 代碼演示

  • 死信隊列設置：
  • 首先須要設置死信隊列的exchange和queue,而後進行綁定： Exchange:dlx.exchange Queue:dlx.queue RoutingKey:#
  • 而後咱們進行正常聲明交換機、隊列、綁定，只不過咱們須要在隊列加上一個參數便可：arguments.put("x-dead-letter-exchange","dlx.exchange");
  • 這樣消息在過時、requeue、隊列在達到最大長度時，消息就能夠直接路由到死信隊列！

7.2.1 生產者

/** * * @ClassName: Producer * @Description: 生產者 * @author Coder編程 * @date2019年7月30日 下午23:15:51 * */
public class Producer {

	
	public static void main(String[] args) throws Exception {
		
		//建立ConnectionFactory
		Connection connection = ConnectionUtils.getConnection();
		Channel channel = connection.createChannel();
		
		String exchange = "test_dlx_exchange";
		String routingKey = "dlx.save";
		
		String msg = "Hello RabbitMQ DLX Message";
		
		for(int i =0; i<1; i ++){
			
			AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
					.deliveryMode(2)
					.contentEncoding("UTF-8")
					.expiration("10000")
					.build();
			channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());
		}
		
	}
}


複製代碼

7.2.2 消費者

/** * * @ClassName: Consumer * @Description: 消費者 * @author Coder編程 * @date2019年7月30日 下午23:13:51 * */
public class Consumer {

	
	public static void main(String[] args) throws Exception {
		
		
		//建立ConnectionFactory
		Connection connection = ConnectionUtils.getConnection();
		Channel channel = connection.createChannel();
		
		// 這就是一個普通的交換機 和 隊列 以及路由
		String exchangeName = "test_dlx_exchange";
		String routingKey = "dlx.#";
		String queueName = "test_dlx_queue";
		
		channel.exchangeDeclare(exchangeName, "topic", true, false, null);
		
		Map<String, Object> agruments = new HashMap<String, Object>();
		agruments.put("x-dead-letter-exchange", "dlx.exchange");
		//這個agruments屬性，要設置到聲明隊列上
		channel.queueDeclare(queueName, true, false, false, agruments);
		channel.queueBind(queueName, exchangeName, routingKey);
		
		//要進行死信隊列的聲明:
		channel.exchangeDeclare("dlx.exchange", "topic", true, false, null);
		channel.queueDeclare("dlx.queue", true, false, false, null);
		channel.queueBind("dlx.queue", "dlx.exchange", "#");
		
		channel.basicConsume(queueName, true, new MyConsumer(channel));
		
		
	}
}


複製代碼

7.2.3 自定義類：MyConsumer

/** * * @ClassName: MyConsumer * @Description: TODO * @author Coder編程 * @date 2019年7月30日 下午23:11:55 * */
public class MyConsumer extends DefaultConsumer {


	public MyConsumer(Channel channel) {
		super(channel);
	}

	@Override
	public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
		System.err.println("-----------consume message----------");
		System.err.println("consumerTag: " + consumerTag);
		System.err.println("envelope: " + envelope);
		System.err.println("properties: " + properties);
		System.err.println("body: " + new String(body));
	}
}

複製代碼

7.2.4 測試結果

運行Consumer，查看管控臺 查看Exchanges

exchange
查看queue
queue
能夠看到test_dlx_queue多了DLX的標識，表示當隊列中出現死信的時候，會將消息發送到死信隊列dlx_queue中

關閉Consumer，只運行Producer

查看queue10秒鐘前

過10秒鐘後，消息過時

查看queue10秒鐘後

在咱們工做中，死信隊列很是重要，用於消息沒有消費者，處於死信狀態。咱們能夠才用補償機制。

小結

本次主要介紹了RabbitMQ的高級特性，首先介紹了互聯網大廠在實際使用中如何保障100%的消息投遞成功和冪等性的，以及對RabbitMQ的確認消息、返回消息、ACK與重回隊列、消息的限流，以及對超時時間、死信隊列的使用

文末

歡迎關注我的微信公衆號：Coder編程 獲取最新原創技術文章和免費學習資料，更有大量精品思惟導圖、面試資料、PMP備考資料等你來領，方便你隨時隨地學習技術知識！ 新建了一個qq羣：315211365，歡迎你們進羣交流一塊兒學習。謝謝了！也能夠介紹給身邊有須要的朋友。

文章收錄至 Github: github.com/CoderMerlin… Gitee: gitee.com/573059382/c… 歡迎關注並star~

微信公衆號

參考文章：

《RabbitMQ消息中間件精講》

推薦文章：

消息中間件——RabbitMQ（五）快速入門生產者與消費者，SpringBoot整合RabbitMQ！

消息中間件——RabbitMQ（六）理解Exchange交換機核心概念!

消息中間件——RabbitMQ（七）高級特性全在這裏!（上）

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程