过期时间

1.单独的设置队列的存活时间,队列中的所有消息的过期时间一样

    @Bean//创建交换机public DirectExchange ttlQueueExchange(){// 交换机名称      是否持久化        是否自动删除return new DirectExchange("ttl_queue_log",true,false);}@Bean//创建队列public Queue ttlQueue(){Map<String, Object> map = new HashMap<>();map.put("x-message-ttl",30000);//表示当前队列的存活时间map.put("x-max-length",1000);//最大容量map.put("x-min-length",0);//最小容量//1.队列名称 2.是否持久化 3.是否唯一绑定某一个交换机  4.是否自动删除 5.给当前的队列配置初始化参数(存活时间,最大容量,最小容量)return new Queue("tt_queue",true,false,false,map);}//将队列和交换机进行绑定@Beanpublic Binding bingCreate(){                                                        //绑定的路由键return BindingBuilder.bind(ttlQueue()).to(ttlQueueExchange()).with("ttl_queue_key");}

   @Test//队列的过期时间public void ttlQueue(){re.convertAndSend("ttl_queue_log","ttl_queue_key","这是过期的消息");}

消息队列是在接收消息的30s才会过期,当然,这个时间也可以在创建队列的时候更改

在队列没有失效之前可以看到队列里的消息,过期后查看就是空数据

2. 对每一条消息单独设置过期时间

    @Beanpublic DirectExchange ttlMessageExchange(){return new DirectExchange("ttl_message_exchange",true,false);}@Beanpublic Queue ttlMessage(){return new Queue("ttl_message",true,false,false,null);}@Beanpublic Binding bindingTtl(){return BindingBuilder.bind(ttlMessage()).to(ttlMessageExchange()).with("mk");}

        注意到这个包:import org.springframework.amqp.core.*;

在发送消息的时候设置消息的过期时间

    @Test//消息的过期时间   消息public void ttlQueueMessage(){MessageProperties messageProperties = new MessageProperties();messageProperties.setExpiration("30000");//设置存活时间 msMessage message = new Message("你好,这是消息的过期时间".getBytes(),messageProperties);re.convertAndSend("ttl_message_exchange","mk",message);}

在设置过期时间内我们可以在ttl_message队列中查看到消息,等超过该时间也会查询不到消息

死信队列

死信队列和普通的队列一样,只是用来独特的信息,比如:被拒绝接收的消息,未被处理的过期消息,超过最大的存储的消息

1.编写核心配置文件application.properties

#开启手动签收(手动ACK)
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 设置消息被拒绝后,不在重新入队
spring.rabbitmq.listener.simple.default-requeue-rejected=false
# 设置消费者需要手动确认消息
spring.rabbitmq.listener.direct.acknowledge-mode=manual

2.创建交换机和队列并绑定

    @Bean//创建一个死信交换机public DirectExchange dealExchange(){return new DirectExchange("dead_letter_exchange",true,false,null);}@Bean//创建死信队列public Queue dealQueue(){return new Queue("dead_letter",true,false,false,null);}@Bean//绑定死信队列和死信消息public Binding bindDead(){return BindingBuilder.bind(dealQueue()).to(dealExchange()).with("dead_log");}@Bean//创建业务层交换机public DirectExchange businessExchange(){return new DirectExchange("business_exchange",true,false,null);}@Bean//普通的队列public Queue testQueue(){HashMap<String, Object> map = new HashMap<>();map.put("x-dead-letter-exchange","dead_letter_exchange");//配置死信交换机map.put("x-dead-letter-routing-key","dead_log");//设置死信交换机和绑定队列之间的路由键return new Queue("test_queue",true,false,false,map);}@Bean//绑定业务处理机和普通队列public Binding testExchange(){return BindingBuilder.bind(testQueue()).to(businessExchange()).with("b_refuse");}@Bean//创建过期消息队列public Queue testttlQueue(){HashMap<String, Object> map = new HashMap<>();map.put("x-dead-letter-exchange","dead_letter_exchange");map.put("x-dead-letter-routing-key","dead_log");map.put("x-message-ttl",30000);return new Queue("test_ttl",true,false,false,map );}@Bean//绑定业务处理机和过期消息队列public Binding testttl(){return BindingBuilder.bind(testttlQueue()).to(businessExchange()).with("d_ttl");}@Bean//溢出队列public Queue testMaxQueue(){HashMap<String, Object> map = new HashMap<>();map.put("x-dead-letter-exchange","dead_letter_exchange");map.put("x-dead-letter-routing-key","dead_log");map.put("x-max-length",3);return new Queue("test_max",true,false,false,map );}@Bean//绑定业务处理机和溢出队列public Binding testMax(){return BindingBuilder.bind(testMaxQueue()).to(businessExchange()).with("t_max");}

3.发送消息

    @Test//死信队列的消息  拒收public void refuseMessage(){re.convertAndSend("business_exchange","b_refuse","死信队列:消息被拒收");}@Test//死信队列的消息   过期public void refuseMessage1(){re.convertAndSend("business_exchange","d_ttl","死信队列:过期的消息");}@Test//死信队列的消息   溢出public void refuseMessage2(){for (int i = 1; i < 6; i++) {re.convertAndSend("business_exchange","t_max","死信队列:溢出的消息"+i);}}

4.处理消息

    @RabbitListener(queues = "test_queue")public void refuseConsumer(String msg,Message messagem,Channel channel) throws IOException {channel.basicNack(messagem.getMessageProperties().getDeliveryTag(),false,false);System.out.println("消息被拒收:"+msg);}

dead_letter死信队列,test_queue拒收队列 , test_ttl过期队列,test_max最大队列

1拒收消息,我们将消息发送到 test_queue队列,但是消息被拒收,消息就会出现在死信队列中,所以我们死信队列中的信息为1,而拒绝队列为0

2.过期消息,在消息没有过期的时候消息在test_ttl队列中,等到有效期结束后就会进入到死信队列

3.消息溢出

延迟队列

在队列中的消息是不需要立即消费的,需要等待一段时间时候才会取出消费,通过死信队列进行中转

1.创建队列并绑定死信队列

    @Bean//创建延迟消息队列public Queue testDelayQueue(){HashMap<String, Object> map = new HashMap<>();map.put("x-dead-letter-exchange","dead_letter_exchange");map.put("x-dead-letter-routing-key","dead_log");map.put("x-message-ttl",30000);return new Queue("test_deal",true,false,false,map );}@Bean//绑定业务处理机和过期延迟队列public Binding testdel(){return BindingBuilder.bind(testDelayQueue()).to(businessExchange()).with("d_deal");}

2.发送消息

    @Test//延迟消息public void dealMessage(){re.convertAndSend("business_exchange","d_deal","延迟的消息:生于小满,小满为安" );}

3.处理消息

    @RabbitListener(queues = "dead_letter")//监听死信队列public void deadLetter(Message message,Channel channel){byte[] body = message.getBody();System.out.println("延迟消息:" +new String(body));}