前言

使用springboot，实现以下功能，有两个队列1、2，往里面发送消息，如果处理失败发生异常，可以重试3次，重试3次均失败，那么就将消息发送到死信队列进行统一处理，例如记录数据库、报警等
完整demo项目代码https://gitee.com/daenmax/rabbit-mq-demoQpJ拜客生活常识网

环境

Windows10，IDEA，otp_win64_25.0，rabbitmq-server-3.10.4
1.双击C:Program FilesRabbitMQ Serverrabbitmq_server-3.10.4sbinrabbitmq-server.bat启动MQ服务
2.然后访问http://localhost:15672/，默认账号密码均为guest，
3.手动添加一个虚拟主机为admin_host，手动创建一个用户账号密码均为adminQpJ拜客生活常识网

pom.xml

复制代码123456XML        
        
            org.springframework.boot
            spring-boot-starter-amqp
            2.7.0
        

配置

复制代码12345678910111213141516YMLspring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: admin
    password: admin
    virtual-host: admin_host
    publisher-confirm-type: correlated
    publisher-returns: true
    listener:
      simple:
        acknowledge-mode: manual
        retry:
          enabled: true    #开启失败重试
          max-attempts: 3    #最大重试次数
          initial-interval: 1000  #重试间隔时间 毫秒

配置文件

RabbitConfigQpJ拜客生活常识网

复制代码123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182JAVApackage com.example.rabitmqdemo.mydemo.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

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


/**
 * Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,
 * Exchange：消息交换机,它指定消息按什么规则,路由到哪个队列。
 * Queue:消息的载体,每个消息都会被投到一个或多个队列。
 * Binding:绑定，它的作用就是把exchange和queue按照路由规则绑定起来.
 * Routing Key:路由关键字,exchange根据这个关键字进行消息投递。
 * vhost:虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。
 * Producer:消息生产者,就是投递消息的程序.
 * Consumer:消息消费者,就是接受消息的程序.
 * Channel:消息通道,在客户端的每个连接里,可建立多个channel.
 */
@Slf4j
@Component
public class RabbitConfig {
    //业务交换机
    public static final String EXCHANGE_PHCP = "phcp";
    //业务队列1
    public static final String QUEUE_COMPANY = "company";
    //业务队列1的key
    public static final String ROUTINGKEY_COMPANY = "companyKey";
    //业务队列2
    public static final String QUEUE_PROJECT = "project";
    //业务队列2的key
    public static final String ROUTINGKEY_PROJECT = "projectKey";

    //死信交换机
    public static final String EXCHANGE_PHCP_DEAD = "phcp_dead";
    //死信队列1
    public static final String QUEUE_COMPANY_DEAD = "company_dead";
    //死信队列2
    public static final String QUEUE_PROJECT_DEAD = "project_dead";
    //死信队列1的key
    public static final String ROUTINGKEY_COMPANY_DEAD = "companyKey_dead";
    //死信队列2的key
    public static final String ROUTINGKEY_PROJECT_DEAD = "projectKey_dead";


//    /**
//     * 解决重复确认报错问题，如果没有报错的话，就不用启用这个
//     *
//     * @param connectionFactory
//     * @return
//     */
//    @Bean
//    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
//        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
//        factory.setConnectionFactory(connectionFactory);
//        factory.setMessageConverter(new Jackson2JsonMessageConverter());
//        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
//        return factory;
//    }

    /**
     * 声明业务交换机
     * 1. 设置交换机类型
     * 2. 将队列绑定到交换机
     * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
     * HeadersExchange ：通过添加属性key-value匹配
     * DirectExchange:按照routingkey分发到指定队列
     * TopicExchange:多关键字匹配
     */
    @Bean("exchangePhcp")
    public DirectExchange exchangePhcp() {
        return new DirectExchange(EXCHANGE_PHCP);
    }

    /**
     * 声明死信交换机
     * 1. 设置交换机类型
     * 2. 将队列绑定到交换机
     * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
     * HeadersExchange ：通过添加属性key-value匹配
     * DirectExchange:按照routingkey分发到指定队列
     * TopicExchange:多关键字匹配
     */
    @Bean("exchangePhcpDead")
    public DirectExchange exchangePhcpDead() {
        return new DirectExchange(EXCHANGE_PHCP_DEAD);
    }

    /**
     * 声明业务队列1
     *
     * @return
     */
    @Bean("queueCompany")
    public Queue queueCompany() {
        Map