1. 安装和配置

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

<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>

1.2 yml 配置

### 生产端的配置
spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: / # 虚拟主机
    username: guest
    password: guest
    publisher-returns: true  #确认消息已经发送到队列,生产上无需开启
    # simple:同步等待confirm结果,直到超时
    #开启消息确认 :correlated:异步回调,MQ返回结果时会回调这个ComfirmCallback
    publisher-confirm-type: correlated #确认消息已发送到交换机

## 生产端的配置
spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: / # 虚拟主机
    username: guest
    password: guest
    publisher-returns: true  #确认消息已经发送到队列,生产上无需开启
    # simple:同步等待confirm结果,直到超时
    #开启消息确认 :correlated:异步回调,MQ返回结果时会回调这个ComfirmCallback
    publisher-confirm-type: correlated #确认消息已发送到交换机

2.生产端的消息确认发送代码

/**
 * (1) RabbitTemplate.ConfirmCallback 这个接口是用来确定消息是否到达交换器的
 * (2) RabbitTemplate.ReturnsCallback 这个则是用来确定消息是否到达队列的，未到达队列时会被调用
 */
@Service
@Slf4j
public class RabbitMqConfirmCallback implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback{

    private RabbitTemplate rabbitTemplate;


    public void queueConfirm(Map<String, String> map) {
        // 第一个参数表示交换机，第二个参数表示 routing key，第三个参数即消息
        rabbitTemplate.convertAndSend("confirm_exchange", "confirm_key1", map, new CorrelationData("111"));
        // 故意输入一个不存在的交换机
        rabbitTemplate.convertAndSend("confirm_exchange_2222", "confirm_key1", map, new CorrelationData("22222"));
        // 故意输入一个不存在的队列
        rabbitTemplate.convertAndSend("confirm_exchange", "confirm_key1_333333", map, new CorrelationData("3333"));
        log.info("Confirm -- 消息--发送结束");
    }


    /**
     * 需要给ConfirmCallback赋值 不然不会走回调方法，默认是null
     * //将当前类的实例设置为 RabbitMQ 的确认回调处理器，跟下面的confirm方法联合使用，
     * // 还需要打开配置：spring: rabbitmq: publisher-confirm-type: correlated
     */
    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    @Autowired
    public RabbitMqConfirmCallback(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
//        rabbitTemplate.setConfirmCallback(this);
    }

 
    /** 此方法用于监听消息是否发送到交换机
     * 回调
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            log.info("confirm -- 监听消息成功发送到交换机--回调id = {}", correlationData);

        } else {
            log.info("confirm -- 消息没有发送到交换机回调id= {},消息发送失败：{}。", correlationData, cause);
        }
    }

    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        log.info("消息未到达队列 --- returnedMessage= " + returnedMessage);
    }
}

2.2 生产端的截图

3.消费端代码

@Component
@Slf4j
public class RabbitConfirmConsumer {

    // 交换机
    public static final String confirm_exchange_name = "confirm_exchange";
    // 队列
    public static final String confirm_queue_name="confirm_queue";
    // routingkey
    public static final String confirm_routing_key = "confirm_key1";

    // 声明交换机
    @Bean("confirmExchange")
    public DirectExchange confirmExchange(){
        return new DirectExchange(confirm_exchange_name);
    }
    // 声明队列
    @Bean("confirmQueue")
    public Queue confirmQueue() {
        return QueueBuilder.durable(confirm_queue_name).build();
    }
    // 绑定队列到交换机
    @Bean
    public Binding queueBingExchange(Queue confirmQueue,DirectExchange confirmExchange){
        return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(confirm_routing_key);
    }


    /**
     * ack：成功处理消息，RabbitMQ从队列中删除该消息
     * nack：消息处理失败，RabbitMQ需要再次投递消息
     * reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息
     */
    @RabbitListener(queues = "confirm_queue")
    public void consumerConfirm(Message message, Channel channel, @Payload Map<String, Object> map,
                                @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        //获取消息的唯一标记
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        log.info("接收的消息为:{},消息的唯一标记={}, 直接注入的tag= {}",message, deliveryTag, tag);
        if(message.getBody() != null){
            //获取消息的内容
            byte[] body = message.getBody();
            //basicAck：表示成功确认，使用此回执方法后，消息会被rabbitmq broker 删除。
            channel.basicAck(deliveryTag,false);//false 表示仅确认当前消息消费成功
            log.info("接收的消息为:{}", map);
        }else{
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
            log.info("未消费数据");
        }
    }

}

3.2消费端截图

4 消费端重试机制

@Service
@Slf4j
public class RabbitRetryConsumer {

    @Bean
    public Queue retryQueue(){
        Map<String,Object> params = new HashMap<>();
        return QueueBuilder.durable("retry_queue").withArguments(params).build();

    }

    @Bean
    public TopicExchange retryTopicExchange(){
        return new TopicExchange("retry_exchange",true,false);
    }
    //队列与交换机进行绑定
    @Bean
    public Binding BindingRetryQueueAndRetryTopicExchange(Queue retryQueue, TopicExchange retryTopicExchange){
        return BindingBuilder.bind(retryQueue).to(retryTopicExchange).with("retry_key");
    }


    int count  = 0;
    //测试重试，需要在yml配置 retry
    @RabbitListener(queues = "retry_queue")
    public void retryConsumer(Map<String, String> map, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        log.info("retryConsumer 重试次数 = {}，重试接收数据为：{}",count++, map);
        int i = 10 /0;
        channel.basicAck(tag,false);
    }

}

4.2 重试机制截图

5. 限流设置--消费端

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual # 开启手动确认模式
        prefetch: 5 #控制消费者从队列中预取(prefetch)消息的数量,以此来实现流控制

5.1 生产端--发送19条信息

@GetMapping("/xianliu")
    public String xianliuTest(){
        for(int i = 1; i < 20; i++){
            Map<String, String> map = new HashMap<>();
            map.put("key","限流测试--" + i);
            rabbitMqProducer.xianliuTest(map);
        }
        return "限流测试发送成功";
    }


/***
     * 限流消息的发送测试
     */
    public void xianliuTest(Map<String, String> map) {
        // 第一个参数表示交换机，第二个参数表示 routing key，第三个参数即消息
        rabbitTemplate.convertAndSend("confirm_exchange", "confirm_key1", map, new CorrelationData("111"));
    }

5.2 消费端

 /**
     * ack：成功处理消息，RabbitMQ从队列中删除该消息
     * nack：消息处理失败，RabbitMQ需要再次投递消息
     * reject：消息处理失败并拒绝该消息，RabbitMQ从队列中删除该消息
     */
    @RabbitListener(queues = "confirm_queue")
    public void consumerConfirm(Message message, Channel channel, @Payload Map<String, Object> map,
                                @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        //获取消息的唯一标记
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        log.info("接收的消息为:{},消息的唯一标记={}, 直接注入的tag= {}",message, deliveryTag, tag);
        if(message.getBody() != null){
            //basicAck：表示成功确认，使用此回执方法后，消息会被rabbitmq broker 删除。
            //channel.basicAck(deliveryTag,false);//false 表示仅确认当前消息消费成功
            log.info("接收的消息为:{}", map);
        }else{
            //否定确认
            //channel.basicNack(deliverTag,false,true);//requeue为false，则变成死信队列
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
            log.info("未消费数据");
        }
    }

5.3 注释掉channel.basicAck--堵塞了

5.4 注释掉了 prefetch -- 19条全部被消费，即使没有ack