1 简介
TTL:
time to live 消息存活时间,如果消息在存活时间内未被消费,则会别清除。 RabbitMQ⽀持两种ttl设置,单独消息进⾏配置ttl,整个队列进⾏配置ttl(居多)。
死信队列:
没有被及时消费的消息存放的队列。
死信交换机:
Dead Letter Exchange(死信交换机,缩写:DLX)当消息成为死信后,会被重新发送到另⼀个交换机,这个交换机就是DLX死信交换机。

消息成为死信的条件:
(1) 消费者拒收消息(basic.reject/ basic.nack),并且没有重新⼊队 requeue=false。
(2) 消息在队列中未被消费,且超过队列或者消息本身的过期时间TTL(time-to-live)。
(3) 队列的消息⻓度达到极限。
消息成为死信后,如果该队列绑定了死信交换机,则消息会被死信交换机重新路由到死信队列。
PS: 死信队列、死信交换机和普通队列、普通交换机没有区别。
2 死信队列代码实现

2.1 pom
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.0.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--日志-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</dependency>
<!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!--公用包-->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.71</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-collections4</artifactId>
<version>4.2</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>2.3.0.RELEASE</version>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.1</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
2.2 application.yml
server:
port: 8080
spring:
rabbitmq:
host: 192.168.38.80
port: 5672
username: guest
password: guest
virtual-host: /
listener:
simple:
#手动确认
acknowledge-mode: manual
2.3 RabbitMqConsts
public final class RabbitMqConsts {
private RabbitMqConsts() {
}
public static final String BUSINESS_EXCHANGE_NAME = "business_exchange";
public static final String BUSINESS_ROUTING_KEY = "business_routing_key";
public static final String DEAD_EXCHANGE_NAME = "dead_exchange";
public static final String BUSINESS_QUEUE = "business_queue";
public static final String DEAD_QUEUE = "dead_queue";
public static final String DEAD_ROUTING_KEY = "dead_routing_key";
}
2.4 DeadRabbitMqConfig
@Configuration
public class DeadRabbitMqConfig {
/**
* 业务交换机
*/
@Bean
public Exchange businessExchange() {
return new DirectExchange(RabbitMqConsts.BUSINESS_EXCHANGE_NAME);
}
/**
* 业务队列
*/
@Bean
public Queue businessQueue() {
//绑定死信交换机
Map<String, Object> paramMap = new HashMap<>();
paramMap.put("x-dead-letter-exchange", RabbitMqConsts.DEAD_EXCHANGE_NAME);
paramMap.put("x-dead-letter-routing-key", RabbitMqConsts.DEAD_ROUTING_KEY);
//当前消息过期时间为5秒
paramMap.put("x-message-ttl", 5000);
return QueueBuilder.durable(RabbitMqConsts.BUSINESS_QUEUE).withArguments(paramMap).build();
}
@Bean
public Binding businessBind(@Qualifier("businessQueue") Queue queue,
@Qualifier("businessExchange") Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConsts.BUSINESS_ROUTING_KEY).noargs();
}
/**
* 死信交换机
*/
@Bean
public Exchange deadExchange() {
return new DirectExchange(RabbitMqConsts.DEAD_EXCHANGE_NAME);
}
/**
* 死信队列
*/
@Bean
public Queue deadQueue() {
return QueueBuilder.durable(RabbitMqConsts.DEAD_QUEUE).build();
}
/**
* 绑定死信交换机和死信队列
*/
@Bean
public Binding deadBing(@Qualifier("deadExchange") Exchange exchange, @Qualifier("deadQueue") Queue queue) {
return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConsts.DEAD_ROUTING_KEY).noargs();
}
}

2.5 RabbitmqConsumer
描述: 死信消费者。
@Component
@RabbitListener(queues = RabbitMqConsts.DEAD_QUEUE)
@Slf4j
public class RabbitmqConsumer {
@RabbitHandler
public void receiveMessage(Message message, String body, Channel channel) throws IOException {
System.out.println("死信队列接收到消息:" + body);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}
2.6 ProviderController
@RestController
@RequestMapping("/provider")
@Slf4j
public class ProviderController {
@Autowired
private RabbitTemplate rabbitTemplate;
@GetMapping("/send")
public String send() {
String message = "Hello SpringBoot RabbitMq";
rabbitTemplate.convertAndSend(RabbitMqConsts.BUSINESS_EXCHANGE_NAME, RabbitMqConsts.BUSINESS_ROUTING_KEY, message );
return "发送成功";
}
}
2.7 启动类
@SpringBootApplication
public class RabbitmqApplication {
public static void main(String[] args) {
SpringApplication.run(RabbitmqApplication.class);
}
}
2.8 测试
结果: 消息发送到业务队列,队列中消息存活时间为5S,超过5S未被消费,消息进入死信队列,消费端监听死信队列消息。



描述: 过5秒后


版权声明:本文为qq_34125999原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接和本声明。