springboot+rabbitmq生产者消息确认机制

springboot+rabbitmq生产者消息确认机制

当发送消息失败时我们希望回调，获取消息内容，rabbitmq提供了这种机制

pom

        
            org.springframework.boot
            spring-boot-starter-web
        
        
        
            com.alibaba
            fastjson
            1.2.75
        
        
            org.springframework.boot
            spring-boot-starter-amqp
        

常量类

package com.example.rabbitmq.constant;

public class Constant {

    //队列名称
    public static final String DIRECT_QUEUE_NAME = "testDirectQueue";

    //交换机名称
    public static final String DIRECT_EXCHANGE_NAME = "testDirectExchange";

    //交换机绑定rout
    public static final String DIRECT_ROUTING = "testDirectRouting";
    //交换机绑定rout
    public static final String DIRECT_ROUTING03 = "testDirectRouting03";
}

回调类         实现/confirm/iCallback接口实现消息发送到交换机的回调

package com.example.rabbitmq.config;

import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;


public class RabbitConfirmCallback implements RabbitTemplate.ConfirmCallback {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            System.out.println("消息发送成功");
        } else {
            System.out.println("消息发送失败");
            System.out.println("失败原因==" + cause);
        }
    }
}

回退回调的实现

package com.example.rabbitmq.config;

import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.nio.charset.StandardCharsets;


public class RabbitReturnCallback implements RabbitTemplate.ReturnsCallback {
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        String msg = new String(returnedMessage.getMessage().getBody(), StandardCharsets.UTF_8);
        System.out.println("发送失败==" + msg);
    }
}

配置类

package com.example.rabbitmq.config;

import com.example.rabbitmq.constant.Constant;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class RabbitConfig {

    @Bean
    public RabbitTemplate initRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //回调类
        rabbitTemplate.setConfirmCallback(new RabbitConfirmCallback());
        //回退回调类
        rabbitTemplate.setReturnsCallback(new RabbitReturnCallback());
        rabbitTemplate.setMandatory(true);
        return rabbitTemplate;
    }

    // 普通队列模式 发送消息的时候指定队列名称
    // 消费者指定队列名称 如果路由模式和普通模式的队列名称一致 消费者相当于消费这两个模式的消息

    // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
    // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
    // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
    //   return new Queue("TestDirectQueue",true,true,false);
    //一般设置一下队列的持久化就好,其余两个就是默认false
    @Bean
    public Queue testDirectQueue() {
        return new Queue(Constant.DIRECT_QUEUE_NAME, true);
    }

    // 路由模式 start
    // 生产者通过路由方式 把队列绑定到交换机上
    // 消息对于多个消费者来说 消息只能被消费一次
    // 生产消息的时候指定交换机和路由，就会发送到绑定的队列上面
    // 消费者只要消费指定的队列就可以了 不会消费同一个交换机上的其他队列
    // 也就是说一个交换机可以绑定多个队列

    //Direct交换机 起名：testDirectExchange
    @Bean
    public DirectExchange testDirectExchange() {
        return new DirectExchange(Constant.DIRECT_EXCHANGE_NAME, true, false);
    }

    //队列和交换机绑定
    @Bean
    public Binding bindingDirect() {
        return BindingBuilder.bind(testDirectQueue()).to(testDirectExchange()).with(Constant.DIRECT_ROUTING);
    }
    // 路由模式 end
}

application.yml

server:
  port: 7001
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: xx
    password: xx
    virtual-host: /
    publisher-/confirm/i-type: correlated

controller类

package com.example.rabbitmq.controller;

import com.alibaba.fastjson.JSON;
import com.example.rabbitmq.constant.Constant;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@RestController
public class RabbitmqController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RequestMapping("/directMq")
    public void directMq() throws Exception{
        Map map = new HashMap<>();
        String id = UUID.randomUUID().toString().replaceAll("-", "");
        map.put("messageId", id);
        map.put("messageData", "hello testDirect");
        map.put("createTime", System.currentTimeMillis());
        CorrelationData correlationData = new CorrelationData(id);
        //Constant.DIRECT_ROUTING 对应的队列是testDirectQueue
        rabbitTemplate.convertAndSend(Constant.DIRECT_EXCHANGE_NAME, Constant.DIRECT_ROUTING, JSON.toJSonString(map), correlationData);
        //Constant.DIRECT_ROUTING03 没有对应的队列
        rabbitTemplate.convertAndSend(Constant.DIRECT_EXCHANGE_NAME, Constant.DIRECT_ROUTING03, JSON.toJSonString(map),correlationData);
    }
}

请求接口测试

 为什么回调两次都是成功，这个暂时还没搞清楚，可以看到没有队列的那个回退回调成功打印【发送失败】的信息