SpringBoot整合RabbitMQ保证消息的可靠的投递及消费

<dependency>
  <groupId>org.springframework.bootgroupId>
  <artifactId>spring-boot-starter-amqpartifactId>
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("消息发送到交换机") ;
        } else {
          System.out.println("消息发送失败 - " + ", 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() {
  // 这里的第二个参数就是设置是否持久化，如果设置为false，当服务重启交换机将丢失
  // 第三个参数是否自动删除，当不再使用该交换机时会自动删除该交换机
  return new TopicExchange("akf.exchange", true, false) ;
}
@Bean
public Queue queue() {
  // 第二个参数true设置队列是持久化的，当服务重启队列不会丢失
  return new Queue("akf.queue", true, false, false) ;
}

Message message = MessageBuilder.withBody("Hello".getBytes())
  // 设置消息投递模式为持久化的（默认不设置就是持久化的）
  .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
  .build() ;

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtualHost: test
    publisherConfirmType: correlated
    publisherReturns: true
    listener:
      simple:
        acknowledgeMode: manual #设置为手动应答

@RabbitListener(queues = {"akf.queue"})
public void onMessage(Message message, Channel channel) throws Exception {
  try {
    System.out.println("接收到消息： " + new String(message.getBody()));
    // ... 这里处理我们的业务代码
    // 当消费者把消息消费成功，再手动应答RabbitMQ
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  } catch (Exception e) {
    // 如果发生了异常，我们一般的处理是直接扔掉死信队列，一般这里出现错误都是消息有问题
    // 如果消息出现问题，你重试再入队列是无意义的
  }
}

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
          # 每次消息重试的间隔乘数
          multiplier: 3
          # 2次间的重试最大间隔时间
          maxInterval: 20000
          maxAttempts: 4 #重试4次，1s, 3s, 9s
          stateless: true #如果消息处理中存在事务则需要将其设置为false

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