004 死信(限制队列最大长度)

文章目录

- 消息ttl过期成为死信
- 队列达到最大长度成为死信
- - MyOrder.java
  - RabbitMQDirectConfig.java
  - OrderProducer.java
  - PayConsumer.java
  - DeadOrderConsumer.java
- application.yaml

死信就是无法被消费的消息。一般来说，producer 将消息投递到 broker 或者直接到 queue 中，consumer 从 queue 取出消息进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

应用场景：
为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。还有比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

死信的原因：

消息 TTL (Time To Live ) : x-message-ttl
队列达到最大长度（队列满了无法再添加数据到 mq 中）： x-max-length
消息被拒绝(basic.reject 或 basic.nack)并且 requeue=false

消息ttl过期成为死信

map.put("x-message-ttl",2000); // 消息存活时间1s

队列达到最大长度成为死信

MyOrder.java


package com.example.direct;

import java.io.Serializable;

public class MyOrder implements Serializable {
    private String orderId;
    private String orderNumber;
    private String customerName;
    private Integer productId;
    private String productName;
    private Float productPrice;
    private Integer productCount;
    private Float orderPrice;

    public MyOrder(){}


    public MyOrder(String orderId, String orderNumber, String customerName, Integer productId, String productName, Float productPrice, Integer productCount, Float orderPrice) {
        this.orderId = orderId;
        this.orderNumber = orderNumber;
        this.customerName = customerName;
        this.productId = productId;
        this.productName = productName;
        this.productPrice = productPrice;
        this.productCount = productCount;
        this.orderPrice = orderPrice;
    }

    public String getOrderId() {
        return orderId;
    }

    public void setOrderId(String orderId) {
        this.orderId = orderId;
    }

    public String getOrderNumber() {
        return orderNumber;
    }

    public Integer getProductId() {
        return productId;
    }

    public void setProductId(Integer productId) {
        this.productId = productId;
    }

    public void setOrderNumber(String orderNumber) {
        this.orderNumber = orderNumber;
    }

    public String getCustomerName() {
        return customerName;
    }

    public void setCustomerName(String customerName) {
        this.customerName = customerName;
    }

    public String getProductName() {
        return productName;
    }

    public void setProductName(String productName) {
        this.productName = productName;
    }

    public Float getProductPrice() {
        return productPrice;
    }

    public void setProductPrice(Float productPrice) {
        this.productPrice = productPrice;
    }

    public Integer getProductCount() {
        return productCount;
    }

    public void setProductCount(Integer productCount) {
        this.productCount = productCount;
    }

    public Float getOrderPrice() {
        return orderPrice;
    }

    public void setOrderPrice(Float orderPrice) {
        this.orderPrice = orderPrice;
    }

    @Override
    public String toString() {
        return "MyOrder{" +
                "orderId=" + orderId +
                ", orderNumber='" + orderNumber + '\'' +
                ", customerName='" + customerName + '\'' +
                ", productName='" + productName + '\'' +
                ", productPrice=" + productPrice +
                ", productCount=" + productCount +
                ", orderPrice=" + orderPrice +
                '}';
    }
}

RabbitMQDirectConfig.java


package com.example.direct;


import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitMQDirectConfig  {

//    1. 创建交换机
//    @Bean
//    public DirectExchange newDirectExchange(){
//        return new DirectExchange("myDirectExchangeAAA",true,false);
//    }

    //2. 创建队列
//    @Bean
//    public Queue newQueueA(){
//        return new Queue("queueAAA",true);
//    }

//3. 绑定队列到交换机中
//    @Bean
//    public Binding bindingA(){
//        return BindingBuilder.bind(newQueueA()).to(newDirectExchange()).with("keyAAA");
//    }








    //==================死信


    //1. 创建交换机
    @Bean
    public DirectExchange newExchange(){
        return new DirectExchange("normalExchange",true,false);
    }
    //2. 创建队列
    @Bean
    public Queue newQueue(){
        Map<String ,Object> map = new HashMap<>();
        //map.put("x-message-ttl",2000); // 消息存活时间1s
        map.put("x-max-length",6); // 队列达到最大长度 为6
        map.put("x-dead-letter-exchange","deadExchange");// 设置死信交换机 的名称
        map.put("x-dead-letter-routing-key","key2") ;//设置死信路由键名字
        return new Queue("normalQueueA",true,false,false,map);

    }
    //3. 绑定
    @Bean
    public Binding binding(){
        return BindingBuilder.bind(newQueue()).to(newExchange()).with("key1");
    }


    //4. 创建死信交换机
    @Bean
    public DirectExchange newDeadExchange(){
        return new DirectExchange("deadExchange",true,false);
    }
    //5. 创建死信队列
    @Bean
    public Queue newDeadQueue(){

        return new Queue("deadQueueA",true,false,false);
    }
    //6. 绑定
    @Bean
    public Binding bindingDead(){
        return BindingBuilder.bind(newDeadQueue()).to(newDeadExchange()).with("key2");
    }








}

OrderProducer.java


package com.example.direct;


import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;

@RestController
@RequestMapping("a")
public class OrderProducer {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    @GetMapping("/submitOrder")
    public String submitOrder(){

        Map<String,Object> map = new HashMap<>();
        map.put("orderNumber","2222");//String
        map.put("productId",1111);//Integer


        for(int i=0;i<=130;i++){

            String orderId = UUID.randomUUID().toString().replace("-","");


            map.put("orderId",orderId);


            rabbitTemplate.convertAndSend("normalExchange", "key1", map);
        }


        return "生产者下单成功";

    }


}

PayConsumer.java



package com.example.direct;


import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

@Component
public class PayConsumer {
    @RabbitHandler
    @RabbitListener(queues = "normalQueueA")
    public void process(Map map, Channel channel, Message message) throws IOException {

        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("支付服务接收到的消息:" + map);
        String orderId = (String)map.get("orderId");//String
        Integer productId = (Integer)map.get("productId");//Integer
        String orderNum = (String)map.get("orderNumber");//String

        System.out.println("支付服务接收到的orderId:" + orderId);
        System.out.println("支付服务接收到的productId:" + productId);
        System.out.println("支付服务接收到的orderNum:" + orderNum);

        //告诉broker，消息已经被确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}

DeadOrderConsumer.java



package com.example.direct;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
public class DeadOrderConsumer {

    // 获得死信队列中的消息
    @RabbitHandler
    @RabbitListener(queues = "deadQueueA")
    public void process(Map map){
        System.out.println("订单取消支付后，从死信队列中接收到的消息:" + map);
        String orderId = (String)map.get("orderId");//String
        Integer productId = (Integer)map.get("productId");//Integer
        String orderNum = (String)map.get("orderNumber");//String


        System.out.println("取消支付后，从死信队列中接收到的orderId:" + orderId);
        System.out.println("取消支付后，从死信队列中接收到的productId:" + productId);
        System.out.println("取消支付后，从死信队列中接收到的orderNum:" + orderNum);


    }
}

application.yaml


server:
  servlet:
    context-path: /app
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    publisher-confirm-type: correlated  # 确认交换机已经接收到生产者的消息了
    publisher-returns: true   #  消息已经到了队列（交换机与队列绑定成功的）
    listener:
      simple:
        acknowledge-mode: manual # 手动消息确认
        concurrency: 1 #消费者数量
        max-concurrency: 1  #消费者最大数量
        prefetch: 1  #消费者每次从队列中取几个消息