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

@Bean
public TopicExchange topicExchange() {
  return new TopicExchange("akf.exchange", true, false) ;
}
@Bean
public Queue queue() {
  return new Queue("akf.queue", true, false, false) ;
}
@Bean
public Binding binding() {
  return BindingBuilder.bind(queue()).to(topicExchange()).with("akf.#") ;
}

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtualHost: test
    publisherConfirmType: correlated  
    publisherReturns: true
    template:
      mandatory: true

@Component
public class ConfigRabbitTemplate implements ApplicationContextAware {

  @Override
  public void setApplicationContext(ApplicationContext context) throws BeansException {
    RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class) ;
    rabbitTemplate.setConfirmCallback(new ConfirmCallback() {
      @Override
      public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("correlation: " + correlationData) ;
        if (ack) {
          System.out.println("消息發(fā)送到交換機") ;
        } else {
          System.out.println("消息發(fā)送失敗 - " + ", cause" + cause) ;
        }
      }
    });
    rabbitTemplate.setReturnsCallback(new ReturnsCallback() {
      @Override
      public void returnedMessage(ReturnedMessage returned) {
        System.out.println(returned.getExchange() + ", " + returned.getRoutingKey() + ", " + returned.getReplyCode() + ", " + returned.getMessage().toString()) ;
      }
    });
  }

}

@Bean
public TopicExchange topicExchange() {
  // 這里的第二個參數(shù)就是設(shè)置是否持久化，如果設(shè)置為false，當(dāng)服務(wù)重啟交換機將丟失
  // 第三個參數(shù)是否自動刪除，當(dāng)不再使用該交換機時會自動刪除該交換機
  return new TopicExchange("akf.exchange", true, false) ;
}
@Bean
public Queue queue() {
  // 第二個參數(shù)true設(shè)置隊列是持久化的，當(dāng)服務(wù)重啟隊列不會丟失
  return new Queue("akf.queue", true, false, false) ;
}

Message message = MessageBuilder.withBody("Hello".getBytes())
  // 設(shè)置消息投遞模式為持久化的（默認(rèn)不設(shè)置就是持久化的）
  .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
  .build() ;

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtualHost: test
    publisherConfirmType: correlated
    publisherReturns: true
    listener:
      simple:
        acknowledgeMode: manual #設(shè)置為手動應(yīng)答

@RabbitListener(queues = {"akf.queue"})
public void onMessage(Message message, Channel channel) throws Exception {
  try {
    System.out.println("接收到消息： " + new String(message.getBody()));
    // ... 這里處理我們的業(yè)務(wù)代碼
    // 當(dāng)消費者把消息消費成功，再手動應(yīng)答RabbitMQ
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  } catch (Exception e) {
    // 如果發(fā)生了異常，我們一般的處理是直接扔掉死信隊列，一般這里出現(xiàn)錯誤都是消息有問題
    // 如果消息出現(xiàn)問題，你重試再入隊列是無意義的
  }
}

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtualHost: test
    publisherConfirmType: correlated
    publisherReturns: true
    listener:
      simple:
        acknowledgeMode: auto
        concurrency: 1
        retry:
          # 開啟重試
          enabled: true
          # 延遲1s后開始重試
          initialInterval: 1000
          # 每次消息重試的間隔乘數(shù)
          multiplier: 3
          # 2次間的重試最大間隔時間
          maxInterval: 20000
          maxAttempts: 4 #重試4次，1s, 3s, 9s
          stateless: true #如果消息處理中存在事務(wù)則需要將其設(shè)置為false

@Bean
public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate) {
  return new RepublishMessageRecoverer(rabbitTemplate, "error.exchange", "error") ;
}