RabbitMQ消息的處理

本文轉載自查看原文 2020-04-19 19:54 612 中間件

Confirm確認與Return返回消息

Confirm消息確認機制：

消息的確認，是指生產者投遞消息后，如果Broker收到消息，則會給我們生產這一個應答。

生產者進行接收應答，用來確定這條消息是否正常的發送到Broker，這種方式也是消息的可靠性投遞的核心保障。

確認機制流程圖：

如何實現Confirm確認消息？

第一步：在channel上開啟確認模式：channel.confirmSelect()

第二步：在channel上添加監聽：addConfirmListener，監聽成功和失敗的返回結果，根據具體的結果對消息進行重新發送、或者記錄日志等后續處理。

消費者：

public class Consumer {
	public static void main(String[] args) throws Exception {
		//1 創建ConnectionFactory
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("101.132.178.134");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		//2 獲取connection
		Connection connection = connectionFactory.newConnection();
		
		//3 通過Connection創建一個新的Channel
		Channel channel = connection.createChannel();
		
		String exchangeName = "test_confirm_exchange";
		String routingKey = "confirm.#";
		String queueName = "test_confirm_queue";
		
		//4 聲明交換機和隊列 然后進行綁定設置, 最后制定路由Key
		channel.exchangeDeclare(exchangeName, "topic", true);
		channel.queueDeclare(queueName, true, false, false, null);
		channel.queueBind(queueName, exchangeName, routingKey);
		
		//5 創建消費者 
		QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
		channel.basicConsume(queueName, true, queueingConsumer);
		
		while(true){
			Delivery delivery = queueingConsumer.nextDelivery();
			String msg = new String(delivery.getBody());
			System.err.println("消費端: " + msg);
		}
	}
}

生產者

public class Producer {
	public static void main(String[] args) throws Exception {

		//1 創建ConnectionFactory
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("101.132.178.134");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		//2 獲取C	onnection
		Connection connection = connectionFactory.newConnection();
		
		//3 通過Connection創建一個新的Channel
		Channel channel = connection.createChannel();

		//4 指定我們的消息投遞模式: 消息的確認模式 
		channel.confirmSelect();
		
		String exchangeName = "test_confirm_exchange";
		String routingKey = "confirm.save";
		
		//5 發送一條消息
		String msg = "Hello RabbitMQ Send confirm message!";
		channel.basicPublish(exchangeName, routingKey, null, msg.getBytes());
		
		//6 添加一個確認監聽
		channel.addConfirmListener(new ConfirmListener() {
			@Override
			public void handleNack(long deliveryTag, boolean multiple) throws IOException {
				System.err.println("-------no ack!-----------");
			}
			@Override
			public void handleAck(long deliveryTag, boolean multiple) throws IOException {
				System.err.println("-------ack!-----------");
			}
		});
	}
}

Return消息機制

Return Listener用於處理一些不可路由的消息

我們的消息生產者，通過指定一個Exchange和Routingkey，把消息送到某一個隊列中，然后我們的消費者監聽隊列，進行消息處理操作。

但是在某些情況下，如果我們在發送消息的時候，當前的exchange不存在或者指定的路由key路由不到，這個時候我們需要監聽這種不可達的消息，就要使用return listener。

在基礎API中有一個關鍵的配置項：

Mandatory：如果為true，則監聽會接收到路由不可達的消息，然后進行后續處理，如果為false，那么broker端自動刪除該消息。（默認false）

public class Consumer {
	public static void main(String[] args) throws Exception {

		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("101.132.178.134");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		Channel channel = connection.createChannel();
		
		String exchangeName = "test_return_exchange";
		String routingKey = "return.#";
		String queueName = "test_return_queue";
		
		channel.exchangeDeclare(exchangeName, "topic", true, false, null);
		channel.queueDeclare(queueName, true, false, false, null);
		channel.queueBind(queueName, exchangeName, routingKey);
		
		QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
		
		channel.basicConsume(queueName, true, queueingConsumer);
		
		while(true){
			Delivery delivery = queueingConsumer.nextDelivery();
			String msg = new String(delivery.getBody());
			System.err.println("消費者: " + msg);
		}
	}
}

public class Producer {
	public static void main(String[] args) throws Exception {
		
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("101.132.178.134");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		Channel channel = connection.createChannel();
		
		String exchange = "test_return_exchange";
		String routingKey = "return.save";
		String routingKeyError = "abc.save";
		
		String msg = "Hello RabbitMQ Return Message";
		
		
		channel.addReturnListener(new ReturnListener() {
			@Override
			public void handleReturn(int replyCode, String replyText, String exchange,
					String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
				
				System.err.println("---------handle  return----------");
				System.err.println("replyCode: " + replyCode);
				System.err.println("replyText: " + replyText);
				System.err.println("exchange: " + exchange);
				System.err.println("routingKey: " + routingKey);
				System.err.println("properties: " + properties);
				System.err.println("body: " + new String(body));
			}
		});
		
		channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());
		//channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());
	}
}

RabbitMQ消費端自定義監聽

我們一般在代碼中編寫while循環，進行consumer.nextDelivery方法進行獲取下一條消息，然后進行消費處理,比較low。
使用自定義的Consumer更加的方便，解耦性更強

自定義Consumer：

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));
	}
}

public class Consumer {
	public static void main(String[] args) throws Exception {
		
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		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);
		
		channel.basicConsume(queueName, true, new MyConsumer(channel));
	}
}

public class Producer {
	public static void main(String[] args) throws Exception {
		
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		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());
		}
	}
}

RabbitMQ消費端限流策略

消費端限流

我們RabbitMQ服務器有上萬條未處理的消息，我們隨便打開一個消費者客戶端，會出現下面情況：巨量的消息瞬間全部推送過來，但是我們單個客戶端無法同時處理這么多數據。（導致服務器崩潰，線上故障）生產端一次推送幾百條數據庫，客戶端只接收一兩條，在高並發的情況下，不能再生產端做限流，只能在消費端處理。

解決方法：

RabbitMQ提供了一種qos（服務質量保證）功能，在非自動確認消息的前提下，如果一定數據的消息（通過基於consumer或者channel設置qos的值）未被確認前，不進行消費新的消息。

void BasicQos（uint prefetchSize，ushort prefetchCount，bool global）；

消費端體現，一次最多能處理多少條消息（基本上為1），限流策略在什么上應用（channel--true，consumer---false）

prefetchSize：0
prefetchCount：會告訴RabbitMQ不要同時給一個消費者推送多余n個消息，一旦有n個消息還沒有ack，則該consumer將block調，知道有消息ack
global：true\false是否將上面設置應用於channel，簡單的說就是上面限制是channel級別的還是consumer級別，基本使用false

prefetchSize和global這兩項，rabbitmq沒有實現，暫不研究
prefetch_count在no_ack=false的情況下生效，在自動應答的情況下兩個值不生效

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));
		channel.basicAck(envelope.getDeliveryTag(), false);
	}
}

public class Consumer {
	public static void main(String[] args) throws Exception {
		
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		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
		channel.basicQos(0, 1, false);
		channel.basicConsume(queueName, false, new MyConsumer(channel));
	}
}

public class Producer {
	public static void main(String[] args) throws Exception {
		
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		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());
		}
		
	}
}

RabbitMQ消費端ACK與重回隊列機制，TTL，死信隊列詳解

消費端的手工ACK和NACK

消費端進行消費的時候，如果由於業務異常我們可以進行日志的記錄，然后進行補償。

如果由於服務器宕機等嚴重問題，那么我們就需要手工進行ACK保障消費端成功。

消費端重回隊列

為了對沒有處理成功的消息，把消息重新回遞給Broker

一般我們在實際應用中，都會關閉重回隊列，也就是設置為false

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) {
			channel.basicNack(envelope.getDeliveryTag(), false, true);
		} else {
			channel.basicAck(envelope.getDeliveryTag(), false);
		}
	}
}

public class Consumer {

	
	public static void main(String[] args) throws Exception {

		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		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));
	}
}

public class Producer {
	public static void main(String[] args) throws Exception {
		
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		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());
		}
	}
}

TTL隊列/消息

TTL是time to live的縮寫,也就是生存時間

RabbitMQ支持消息的過期時間，在消息發送時可以進行指定

RabbitMQ支持隊列的過期時間，從消息入隊列開始計算，只要超過了隊列的超過時間配置，那么消息會自動的清除

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

消息10s過期，TTL是隊列過期時間

DLX死信隊列

DLX，Dead-Letter-Exchange

利用DLX，當消息在一個隊列中變成死信之后，它能夠被重新publish到另一個exchange，這個exchange就是DLX。

消息變成死信情況：

消息被拒絕（basic.reject/basic.nack）並且request=false
消息TTL過期
隊列達到最大的長度

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

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

可以監聽這個隊列中消息做相應的處理，這個特性可以彌補RabbitMQ3.0以前支持的immediate參數的功能。　

死信隊列設置：

首先要設置死信隊列的exchange和queue，並進行綁定，然后我們進行正常聲明交換機，隊列，綁定，只不過我們需要在隊列加上一個參數：arguments.put("x-dead-letter-exchange","dlx.exchange");

這樣消息在過期、request、隊列達到最大長度時，消息就可以直接路由到死信隊列。　

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));
	}
}

public class Consumer {
	public static void main(String[] args) throws Exception {
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		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));
	}
}

public class Producer {
	public static void main(String[] args) throws Exception {
		
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		Connection connection = connectionFactory.newConnection();
		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());
		}
	}
}

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 RabbitMQ-處理消息丟失 RabbitMQ（六）消息冪等性處理 RabbitMQ的消息持久化處理 RabbitMQ中各種消息類型如何處理？ RabbitMQ中各種消息類型如何處理如何處理RabbitMQ 消息堆積和消息丟失問題 rabbitmq系列（三）消息冪等性處理 RabbitMQ(五)消息發送失敗后的處理 C# 封裝RabbitMQ消息隊列處理 SpringBoot+RabbitMQ學習筆記（五）RabbitMQ消息持久化處理