<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  rabbitmq:
    host: 10.49.196.10
    port: 5672
    virtual-host: /
    username: guest
    password: guest
    publisher-confirm-type: simple #消息发送确认模式 none、correlated、simple；Confirm模式只管有无投递到exchange，而不管有无发送到队列当中
    publisher-returns: true #当消息未投递到queue时消息退是否回

    template:
      mandatory: true #消息在没有被队列接收时是否退回，与spring.rabbitmq.publisher-returns类似，该配置优先级高于spring.rabbitmq.publisher-returns
    listener:
      type: simple #容器类型
      simple:
        acknowledge-mode: manual #消息消费确认模式

package com.abc.demo.general.rabbit.springboot;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    public static String EXCHANGE_NAME = "exchange.log";
    public static final String QUEUE_NAME = "queue.file";

    /**
     * 声明direct类型交换机

     * @return
     */
    @Bean("exchangeLog")
    public Exchange exchange() {
        return ExchangeBuilder.directExchange(EXCHANGE_NAME).durable(true).build();
    }

    /**
     * 声明队列
     * @return
     */
    @Bean("queueFile")
    public Queue queue(){
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    /**
     * 绑定队列和交换机
     * @param queue
     * @param exchange
     * @return
     */
    @Bean
    public Binding bindingWarn(@Qualifier("queueFile") Queue queue, @Qualifier("exchangeLog") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("warn").noargs();
    }

    @Bean
    public Binding bindingInfo(@Qualifier("queueFile") Queue queue, @Qualifier("exchangeLog") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("info").noargs();
    }
}

package com.abc.demo.general.rabbit.springboot;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class Producer implements InitializingBean {
    private static Logger logger = LoggerFactory.getLogger(Producer.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private static int index = 0;

    @Scheduled(cron = "0/5 * * * * *")
    public void sendMsg() {
        String msg = "消息-" + index++;
        logger.info("发送消息:{}", msg);
        String routingKey = "";
        if (index % 3 == 0) {
            routingKey = "warn";
        } else if(index % 3 == 1) {
            routingKey = "info";
        }  else {
            routingKey = "debug";
        }

        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, routingKey, msg);
    }

    /**
     * 发送消息确认

     * spring.rabbitmq.publisher-confirm-type=correlated
     */
    //@Scheduled(cron = "0/5 * * * * *")
    public void sendMsgConfirmCorrelated() {
        index++;
        String msg = "消息-" + index;
        logger.info("发送消息:{}", msg);
        String routingKey = "";
        if (index % 3 == 0) {
            routingKey = "warn";
        } else if(index % 3 == 1) {
            routingKey = "info";
        }  else {
            routingKey = "debug";
        }
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                logger.info("消息:{},发送成功。", correlationData);
            } else {
                logger.info("消息:{},发送失败。原因：{}", correlationData, cause);
            }
        });

        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(index + "");
        //用不存在的exchange测试
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME + "2", routingKey, msg, correlationData);
    }

    /**
     * 发送消息确认

     * spring.rabbitmq.publisher-confirm-type=simple
     */
    //@Scheduled(cron = "0/5 * * * * *")
    public void sendMsgConfirmSimple() {
        index++;
        String msg = "消息-" + index;
        logger.info("发送消息:{}", msg);
        String routingKey = "";
        if (index % 3 == 0) {
            routingKey = "warn";
        } else if(index % 3 == 1) {
            routingKey = "info";
        }  else {
            routingKey = "debug";
        }

        final String rk = routingKey;
        rabbitTemplate.invoke(operations -> {
            rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME, rk, msg);
            //rabbitTemplate.waitForConfirmsOrDie(2000);
            boolean result = rabbitTemplate.waitForConfirms(2000);
            if (result) {
                logger.info("消息发送成功！");
            } else {
                logger.info("消息发送失败！");
            }
            return null;
        });
    }

    @Override
    public void afterPropertiesSet() {
        /**
         * 消息未发送到queue反馈，可以用不存在的routingKey来测试

         * spring.rabbitmq.publisher-returns=true
         */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            logger.warn("返回消息配置：{}", message.getMessageProperties().toString());
            logger.warn("反馈代码：{}", replyCode);
            logger.warn("反馈内容：{}", replyText);
            logger.warn("exchange：{}", exchange);
            logger.warn("routingKey：{}", routingKey);
        });
    }
}

package com.abc.demo.general.rabbit.springboot;

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

import java.io.IOException;

@Component
public class Consumer {
    private static Logger logger = LoggerFactory.getLogger(Consumer.class);

    @RabbitListener(queues = RabbitMQConfig.QUEUE_NAME)
    public void process(Message message, Channel channel) throws IOException {
        try {
            logger.info("接受到消息：" + new String(message.getBody()));
            channel.basicQos(1);//业务处理...

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}