RabbitMQ：高级特性 - 版本历史

Eijux：/* 优先级队列（消息优先级） */

2022-10-31T13:17:29Z

优先级队列（消息优先级）

←上一版本		2022年10月31日 (一) 21:17的版本
第76行：		第76行：
	</syntaxhighlight>		</syntaxhighlight>
	#* 只能通过声明方式设定，不能通过策略方式修改；		#* 只能通过声明方式设定，不能通过策略方式修改；
	# '''设置消息优先级'''~~：通过消息属性（Priority）~~		# '''设置消息优先级'''：通过消息属性（'''Priority'''）
	#: <syntaxhighlight lang="Java" highlight="">		#: <syntaxhighlight lang="Java" highlight="">
	for(int i=0;i<10;i++) {		for(int i=0;i<10;i++) {
第90行：		第90行：
	@Override		@Override
	public Message postProcessMessage(Message message) throws AmqpException {		public Message postProcessMessage(Message message) throws AmqpException {
	message.getMessageProperties().setPriority(~~finalI~~);		message.getMessageProperties().setPriority(final1);
	return message;		return message;
	}		}
第97行：		第97行：


	'''优先级队列必须和优先级消息一起使用'''，才能发挥出效果，但是会消耗性能。		'''优先级队列必须和优先级消息一起使用'''，才能发挥出效果，但是会消耗性能。

	== 死信队列 ==		== 死信队列 ==

Eijux：/* 消费消息的顺序 */

2022-10-31T13:13:48Z

消费消息的顺序

←上一版本		2022年10月31日 (一) 21:13的版本
第45行：		第45行：
	: 例如：消息A、B、C按顺序进入队列，消费者A1拿到消息A、消费者B1拿到消息B, 结果消费者B执行速度快，就跑完了，又或者消费者A1挂了，都会导致消息顺序不一致。		: 例如：消息A、B、C按顺序进入队列，消费者A1拿到消息A、消费者B1拿到消息B, 结果消费者B执行速度快，就跑完了，又或者消费者A1挂了，都会导致消息顺序不一致。

	'''解决消费顺序的问题，通常就是一个队列只有一个消费者'''。		'''解决消费顺序的问题，通常就是一个队列只有一个消费者'''。
	* 这样可以按顺序处理消息，但并发能力下降了，无法并发消费消息，这是个取舍问题。		* 这样可以按顺序处理消息，但并发能力下降了，无法并发消费消息，这是个取舍问题。
	* 如果业务又要顺序消费，又要增加并发，通常思路就是开启多个队列，业务根据规则将消息分发到不同的队列，通过增加队列的数量来提高并发度。		* 如果业务又要顺序消费，又要增加并发，通常思路就是开启多个队列，业务根据规则将消息分发到不同的队列，通过增加队列的数量来提高并发度。

Eijux：/* 消息顺序 */

2022-10-31T13:12:15Z

消息顺序

←上一版本		2022年10月31日 (一) 21:12的版本
第28行：		第28行：
	RabbitMQ使用过程中，有些业务场景需要我们保证顺序消费，例如：业务上产生三条消息，分别是对数据的增加、修改、删除操作，如果没有保证顺序消费，执行顺序可能变成删除、修改、增加，这就乱了。		RabbitMQ使用过程中，有些业务场景需要我们保证顺序消费，例如：业务上产生三条消息，分别是对数据的增加、修改、删除操作，如果没有保证顺序消费，执行顺序可能变成删除、修改、增加，这就乱了。

	: [[File:RabbitMQ：消息顺序示例.png\|~~400px~~]]		: [[File:RabbitMQ：消息顺序示例.png\|800px]]

Eijux：/* 消息过期时间（TTL） */

2021-05-27T11:01:26Z

消息过期时间（TTL）

←上一版本		2021年5月27日 (四) 19:01的版本
第239行：		第239行：
	#: <syntaxhighlight lang="Java" highlight="">		#: <syntaxhighlight lang="Java" highlight="">
	Map<String, Object> arguments = new HashMap<String, Object>();		Map<String, Object> arguments = new HashMap<String, Object>();
	arguments.put("x-~~dead~~-~~letter-exchange~~", "~~dlx.exchange~~"); // ~~设置 x-dead-letter-exchange 属性为：dlx.exchange（死信交换机名称）~~		arguments.put("x-message-ttl", "60000"); // 设置队列消息的过期时间（单位 ms）

	channel.queueDeclare(queueName, true, false, false, arguments);		channel.queueDeclare(queueName, true, false, false, arguments);

Eijux：/* 处理步骤 */

2021-05-27T10:56:44Z

处理步骤

←上一版本		2021年5月27日 (四) 18:56的版本
第159行：		第159行：
	String msg = "this is dlx msg";		String msg = "this is dlx msg";

	// 设置消息过期时间，10秒后再消费让消息进入死信队列（死信条件）		// 设置消息过期时间，10秒后再消费让消息进入死信队列（死信条件）
	AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()		AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
	.deliveryMode(2)		.deliveryMode(2)
	.expiration("10000")		.expiration("10000")
第186行：		第186行：
	String routingKey = "item.#";		String routingKey = "item.#";

	// 声明交换机		// 声明交换机
	channel.exchangeDeclare(exchangeName, "topic", true, false, null);		channel.exchangeDeclare(exchangeName, "topic", true, false, null);
	// 声明队列		// 声明队列
	// 1、设置参数到 arguments 中		// 1、设置参数到 arguments 中
	Map<String, Object> arguments = new HashMap<String, Object>();		Map<String, Object> arguments = new HashMap<String, Object>();
	arguments.put("x-dead-letter-exchange", "dlx.exchange"); // ~~设置 x-dead-letter-~~exchange ~~属性为：dlx.exchange（死信交换机名称）~~		arguments.put("x-dead-letter-exchange", "dlx.exchange"); // 设置死信交换机名称为：dlx.exchange
	// 2、将 arguments 放入队列的声明中		// 2、将 arguments 放入队列的声明中
	channel.queueDeclare(queueName, true, false, false, arguments);		channel.queueDeclare(queueName, true, false, false, arguments);
	// 绑定		// 绑定
第198行：		第198行：


	// 声明死信交换机		// 声明死信交换机
	channel.exchangeDeclare("dlx.exchange", "topic", true, false, null);		channel.exchangeDeclare("dlx.exchange", "topic", true, false, null);
	// 声明死信队列		// 声明死信队列
	channel.queueDeclare("dlx.queue", true, false, false, null);		channel.queueDeclare("dlx.queue", true, false, false, null);
第206行：		第206行：


	// 消费者		// 消费者
	Consumer consumer = new DefaultConsumer(channel) {		Consumer consumer = new DefaultConsumer(channel) {
	@Override		@Override
	public void handleDelivery(String consumerTag, Envelope envelope,		public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
	AMQP.BasicProperties properties, byte[] body)
	throws IOException {		throws IOException {
	String message = new String(body, "UTF-8");		String message = new String(body, "UTF-8");
第220行：		第219行：


	// 死信消费者		// 死信消费者
	Consumer dlxConsumer = new DefaultConsumer(channel) {		Consumer dlxConsumer = new DefaultConsumer(channel) {
	@Override		@Override
	public void handleDelivery(String consumerTag, Envelope envelope,		public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
	AMQP.BasicProperties properties, byte[] body)
	throws IOException {		throws IOException {
	String message = new String(body, "UTF-8");		String message = new String(body, "UTF-8");

Eijux：/* 延迟消息插件 */

2021-05-27T10:47:31Z

延迟消息插件

←上一版本		2021年5月27日 (四) 18:47的版本
第288行：		第288行：
	rabbitmq-plugins enable rabbitmq_delayed_message_exchange		rabbitmq-plugins enable rabbitmq_delayed_message_exchange
	</syntaxhighlight>		</syntaxhighlight>
	#: [[File:RabbitMQ：安装插件rabbitmq_delayed_message_exchange.png\|~~400px~~]]		#: [[File:RabbitMQ：安装插件rabbitmq_delayed_message_exchange.png\|800px]]
	#* 安装成功后，重启 RabbitMQ；		#* 安装成功后，重启 RabbitMQ；
	# '''定义交换机'''：		# '''定义交换机'''：

Eijux：/* 消息过期时间（TTL） */

2021-05-27T10:46:38Z

消息过期时间（TTL）

显示更改

Eijux：/* 死信处理步骤 */

2021-05-27T10:05:12Z

死信处理步骤

←上一版本		2021年5月27日 (四) 18:05的版本
第138行：		第138行：
	</syntaxhighlight>		</syntaxhighlight>

	=== ~~死信处理步骤~~ ===		=== 处理步骤 ===
	# '''定义死信交换机'''；		# '''定义死信交换机'''；
	# '''定义死信队列'''；		# '''定义死信队列'''；

Eijux：/* 死信产生的条件 */

2021-05-27T10:05:00Z

死信产生的条件

←上一版本		2021年5月27日 (四) 18:05的版本
第106行：		第106行：


	=== ~~死信产生的条件~~ ===		=== 产生条件 ===
	# 消息被消费者手动拒绝（basic.reject / basic.nack），并且“requeue = false”（禁止重新入队）；		# 消息被消费者手动拒绝（basic.reject / basic.nack），并且“requeue = false”（禁止重新入队）；
	#: <syntaxhighlight lang="Java" highlight="">		#: <syntaxhighlight lang="Java" highlight="">
第137行：		第137行：
	}		}
	</syntaxhighlight>		</syntaxhighlight>


	=== 死信处理步骤 ===		=== 死信处理步骤 ===

Eijux：/* 死信队列 */

2021-05-27T10:04:44Z

死信队列

←上一版本		2021年5月27日 (四) 18:04的版本
第106行：		第106行：


	~~死信产生的条件：~~		=== 死信产生的条件 ===
	# 消息被消费者手动拒绝（basic.reject / basic.nack），并且“requeue = false”（禁止重新入队）；		# 消息被消费者手动拒绝（basic.reject / basic.nack），并且“requeue = false”（禁止重新入队）；
	#: <syntaxhighlight lang="Java" highlight="">		#: <syntaxhighlight lang="Java" highlight="">
第139行：		第139行：


	~~处理步骤：~~		=== 死信处理步骤 ===
	# '''定义死信交换机'''；		# '''定义死信交换机'''；
	# '''定义死信队列'''；		# '''定义死信队列'''；

RabbitMQ：高级特性 - 版本历史

Eijux：​/* 优先级队列（消息优先级） */

Eijux：​/* 消费消息的顺序 */

Eijux：​/* 消息顺序 */

Eijux：​/* 消息过期时间（TTL） */

Eijux：​/* 处理步骤 */

Eijux：​/* 延迟消息插件 */

Eijux：​/* 消息过期时间（TTL） */

Eijux：​/* 死信处理步骤 */

Eijux：​/* 死信产生的条件 */

Eijux：​/* 死信队列 */

Eijux：/* 优先级队列（消息优先级） */

Eijux：/* 消费消息的顺序 */

Eijux：/* 消息顺序 */

Eijux：/* 消息过期时间（TTL） */

Eijux：/* 处理步骤 */

Eijux：/* 延迟消息插件 */

Eijux：/* 消息过期时间（TTL） */

Eijux：/* 死信处理步骤 */

Eijux：/* 死信产生的条件 */

Eijux：/* 死信队列 */