1、RabbitMQ死信队列2、代码示例2.1、队列过期2.1.1、配置类RabbitConfig关键代码2.1.2、业务类MessageService2.1.3、配置文件application.yml2.1.4、启动类2.1.5、配置文件2.1.6、测试

2.2、消息过期2.2.1、配置类RabbitConfig2.2.2、业务类MessageService关键代码2.2.3、配置文件application.yml2.2.4、启动类同上2.2.5、配置文件同上2.2.6、测试

2.3、队列达到最大长度先入队的消息会被发送到DLX2.3.1、配置类RabbitConfig关键代码2.3.2、业务类MessageService关键代码2.3.3、配置文件application.yml2.3.4、启动类同上2.3.5、配置文件pom.xml同上2.3.6、测试

2.4、消费者拒绝消息不进行重新投递2.4.1、生产者2.4.1.1、生产者application.yml2.4.1.2、生产者发送消息2.4.1.3、生产者配置类

2.4.2、消费者2.4.2.1、消费者application.yml

启动手动确认关键配置

org.springframework.amqp.core.*;

import

org.springframework.beans.factory.annotation.Value;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;import

java.util.HashMap;

{Value(${my.exchangeNormalName})private

String

exchangeNormalName;Value(${my.queueNormalName})private

String

queueNormalName;Value(${my.exchangeDlxName})private

String

exchangeDlxName;Value(${my.queueDlxName})private

String

ExchangeBuilder.directExchange(exchangeNormalName).build();}/***

正常队列*

HashMap();arguments.put(x-message-ttl,20000);//设置队列的过期时间为20秒//重点设置这两个参数arguments.put(x-dead-letter-exchange,exchangeDlxName);

//设置队列的死信交换机arguments.put(x-dead-letter-routing-key,error);//设置死信路由key要跟死信交换机和死信队列绑定的路由key一致return

QueueBuilder.durable(queueNormalName).withArguments(arguments)

正常交换机和正常队列绑定*

BindingBuilder.bind(normalQueue).to(normalExchange).with(order);}/***

死信交换机*

ExchangeBuilder.directExchange(exchangeDlxName).build();}/***

死信队列*

QueueBuilder.durable(queueDlxName).build();}/***

死信交换机和死信队列绑定*

BindingBuilder.bind(dlxQueue).to(dlxExchange).with(error);}

package

org.springframework.amqp.core.Message;

import

org.springframework.amqp.core.MessageBuilder;

import

org.springframework.amqp.rabbit.core.RabbitTemplate;

import

org.springframework.context.annotation.Bean;

import

org.springframework.stereotype.Service;import

import

world.getBytes()).build();rabbitTemplate.convertAndSend(exchange.normal.a,order,message);log.info(消息发送完毕发送时间是new

Date());}

#正常队列没有消费组设置过期时间exchangeDlxName:

exchange.dlx.a

com.power.service.MessageService;

import

org.springframework.boot.ApplicationArguments;

import

org.springframework.boot.ApplicationRunner;

import

org.springframework.boot.SpringApplication;

import

org.springframework.boot.autoconfigure.SpringBootApplication;import

javax.annotation.Resource;SpringBootApplication

public

{SpringApplication.run(Application.class);}Overridepublic

void

xmlnshttp://maven.apache.org/POM/4.0.0

xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0

http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcom.power/groupIdartifactIdrabbit_06_dlx01/artifactIdversion1.0-SNAPSHOT/versionnamerabbit_06_dlx01/namepropertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.targetproject.build.sourceEncodingUTF-8/project.build.sourceEncoding/propertiesparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.6.13/versionrelativePath//parentdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.24/version/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project2.1.6、测试

启动程序发送消息

消息会先被发送到正常队列queue.normal.a中超时未被消费

package

org.springframework.amqp.core.*;

import

org.springframework.beans.factory.annotation.Value;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;import

java.util.HashMap;

{Value(${my.exchangeNormalName})private

String

exchangeNormalName;Value(${my.queueNormalName})private

String

queueNormalName;Value(${my.exchangeDlxName})private

String

exchangeDlxName;Value(${my.queueDlxName})private

String

ExchangeBuilder.directExchange(exchangeNormalName).build();}/***

正常队列*

HashMap();//重点设置这两个参数//设置队列的死信交换机arguments.put(x-dead-letter-exchange,exchangeDlxName);//设置死信路由key要跟死信交换机和死信队列绑定的路由key一致arguments.put(x-dead-letter-routing-key,error);return

QueueBuilder.durable(queueNormalName).withArguments(arguments)

正常交换机和正常队列绑定*

BindingBuilder.bind(normalQueue).to(normalExchange).with(order);}/***

死信交换机*

ExchangeBuilder.directExchange(exchangeDlxName).build();}/***

死信队列*

QueueBuilder.durable(queueDlxName).build();}/***

死信交换机和死信队列绑定*

BindingBuilder.bind(dlxQueue).to(dlxExchange).with(error);}

package

org.springframework.amqp.core.Message;

import

org.springframework.amqp.core.MessageBuilder;

import

org.springframework.amqp.core.MessageProperties;

import

org.springframework.amqp.rabbit.core.RabbitTemplate;

import

org.springframework.context.annotation.Bean;

import

org.springframework.stereotype.Service;import

import

MessageProperties();//设置单条消息的过期时间单位为毫秒数据类型为字符串messageProperties.setExpiration(20000);Message

message

world.getBytes()).andProperties(messageProperties).build();rabbitTemplate.convertAndSend(exchange.normal.02,order,message);}catch

(Exception

e){e.printStackTrace();log.info(消息发送失败new

Date());}log.info(消息发送完毕发送时间是new

Date());}

#正常队列没有消费组设置过期时间exchangeDlxName:

exchange.dlx.02

org.springframework.amqp.core.*;

import

org.springframework.beans.factory.annotation.Value;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;import

java.util.HashMap;

{Value(${my.exchangeNormalName})private

String

exchangeNormalName;Value(${my.queueNormalName})private

String

queueNormalName;Value(${my.exchangeDlxName})private

String

exchangeDlxName;Value(${my.queueDlxName})private

String

ExchangeBuilder.directExchange(exchangeNormalName).build();}/***

正常队列*

HashMap();//设置队列的最大长度arguments.put(x-max-length,5);//重点设置这两个参数//设置队列的死信交换机arguments.put(x-dead-letter-exchange,exchangeDlxName);//设置死信路由key要跟死信交换机和死信队列绑定的路由key一致arguments.put(x-dead-letter-routing-key,error);return

QueueBuilder.durable(queueNormalName).withArguments(arguments)

正常交换机和正常队列绑定*

BindingBuilder.bind(normalQueue).to(normalExchange).with(order);}/***

死信交换机*

ExchangeBuilder.directExchange(exchangeDlxName).build();}/***

死信队列*

QueueBuilder.durable(queueDlxName).build();}/***

死信交换机和死信队列绑定*

BindingBuilder.bind(dlxQueue).to(dlxExchange).with(error);}

package

org.springframework.amqp.core.Message;

import

org.springframework.amqp.core.MessageBuilder;

import

org.springframework.amqp.rabbit.core.RabbitTemplate;

import

org.springframework.context.annotation.Bean;

import

org.springframework.stereotype.Service;import

import

MessageBuilder.withBody(msg.getBytes()).build();rabbitTemplate.convertAndSend(exchange.normal.03,order,message);log.info(消息发送完毕发送时间是new

Date());}}

#正常队列没有消费组设置过期时间exchangeDlxName:

exchange.dlx.03

消费者从正常的队列接收消息但是消费者对消息不进行确认并且不对消息进行重新投递此时消息就进入死信队列。

2.4.1、生产者

#正常队列没有消费组设置过期时间exchangeDlxName:

exchange.dlx.04

org.springframework.amqp.core.Message;

import

org.springframework.amqp.core.MessageBuilder;

import

org.springframework.amqp.rabbit.core.RabbitTemplate;

import

org.springframework.context.annotation.Bean;

import

org.springframework.stereotype.Service;import

import

MessageBuilder.withBody(msg.getBytes()).build();rabbitTemplate.convertAndSend(exchange.normal.04,order,message);log.info(消息发送完毕发送时间是new

Date());}

org.springframework.amqp.core.*;

import

org.springframework.beans.factory.annotation.Value;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;import

java.util.HashMap;

{Value(${my.exchangeNormalName})private

String

exchangeNormalName;Value(${my.queueNormalName})private

String

queueNormalName;Value(${my.exchangeDlxName})private

String

exchangeDlxName;Value(${my.queueDlxName})private

String

ExchangeBuilder.directExchange(exchangeNormalName).build();}/***

正常队列*

HashMap();//重点设置这两个参数//设置队列的死信交换机arguments.put(x-dead-letter-exchange,exchangeDlxName);//设置死信路由key要跟死信交换机和死信队列绑定的路由key一致arguments.put(x-dead-letter-routing-key,error);return

QueueBuilder.durable(queueNormalName).withArguments(arguments)

正常交换机和正常队列绑定*

BindingBuilder.bind(normalQueue).to(normalExchange).with(order);}/***

死信交换机*

ExchangeBuilder.directExchange(exchangeDlxName).build();}/***

死信队列*

QueueBuilder.durable(queueDlxName).build();}/***

死信交换机和死信队列绑定*

BindingBuilder.bind(dlxQueue).to(dlxExchange).with(error);}

}2.4.2、消费者

powerlistener:simple:acknowledge-mode:

manual关键配置

org.springframework.amqp.core.Message;

import

org.springframework.amqp.core.MessageProperties;

import

org.springframework.amqp.rabbit.annotation.RabbitListener;

import

org.springframework.stereotype.Component;import

java.io.IOException;

{RabbitListener(queues{queue.normal.04})public

void

channel){//获取消息属性MessageProperties

messageProperties

message.getMessageProperties();//获取消息的唯一标识类似学号和身份证号long

deliveryTag

messageProperties.getDeliveryTag();try{byte[]

body

String(body);log.info(监听到的消息是msg,接收的时间是new

Date());//TODO

a1/0;//消费者的手动确认false只确认当前消息true批量确认channel.basicAck(deliveryTag,false);}catch

(Exception

e){log.error(接收者出现问题{},e.getMessage());try

channel.basicNack(deliveryTag,false,true);//消费者的手动不确认参数3false

不重新入队不重新投递就会变成死信channel.basicNack(deliveryTag,false,false);}catch

(IOException