springboot中使用rabbitmq

1.加入依赖

    org.springframework.boot
    spring-boot-starter-amqp

 2. 配置，配置中各个参数的含义 传送门

spring:
    rabbitmq:
      host: xxxx
      username: xxx
      password: xxxx
      port: 5672
      virtual-host: /
      publisher-confirm-type: correlated
      publisher-returns: true
      template:
        mandatory: true
      listener:
        type: simple
        simple:
          acknowledge-mode: manual
          retry:
            enabled: true
          prefetch: 30

3.使用，我这里是根据自己的业务场景的具体使用，可以看这个大神总结的使用方式传送门 

3.1 配置一个topic类型的交换机，绑定队列，指定routingkey 

@Configuration
public class TopicRabbitMqConfig {
 
    public final static String exchange = "xxx";
    public final static String queue = "xxx";
    private final static String routing = "xxx";
 
    @Bean
    TopicExchange netdiskTopicExchange(){
        return new TopicExchange(exchange, true, false);
    }
 
    @Bean
    Queue netdiskQueue(){
        return new Queue(queue);
    }
 
    @Bean
    Binding netdiskBinding(){
        return BindingBuilder.bind(netdiskQueue()).to(netdiskTopicExchange()).with(routing);
    }
}

3.2 封装了一个工具类方便后续使用

@Slf4j
@Component
public class MqUtil implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
 
    @Resource
    private RabbitTemplate rt;
 
    public static RabbitTemplate rabbitTemplate;
 
    @PostConstruct
    private void init() {
        MqUtil.rabbitTemplate = this.rt;
 
        rabbitTemplate.setConfirmCallback(this::confirm);
        rabbitTemplate.setReturnsCallback(this::returnedMessage);
    }
 
 
 
    /**
     * 不论是否进入交换机，都会回调当前方法
     *
     * @param correlationData 消息投递封装对象
     * @param ack             是否投递成功
     * @param exception       如果错误，错误原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String exception) {
 
        if (!ack) {
            if (correlationData instanceof CorrelationDataExt) {
                CorrelationDataExt correlationDataExt = (CorrelationDataExt) correlationData;
                Object message = correlationDataExt.getData();
                log.error("消息进入交换机失败：{}, 原因：{}", JSON.toJSONString(message), exception);
            }
        }
 
    }
 
    /**
     * 消息从交换机进入队列失败回调方法：只会在失败的情况下
     *
     * @param ReturnedMessage returnedMessage
     */
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        Message message = returnedMessage.getMessage();
        int replyCode = returnedMessage.getReplyCode();
        String replyText = returnedMessage.getReplyText();
        String exchange = returnedMessage.getExchange();
        String routingKey = returnedMessage.getRoutingKey();
        String messageContent = new String(message.getBody(), StandardCharsets.UTF_8);
 
        Map map = new HashMap<>();
        map.put("replyCode", replyCode);
        map.put("replyText", replyText);
        map.put("exchange", exchange);
        map.put("routingKey", routingKey);
        map.put("message", messageContent);
 
        log.error("消息从交换机进入队列失败：{}", JSON.toJSONString(map));
    }
 
 
 
 
    public static void send(String type, MqMessageData data) throws AmqpException {
        String msgId = UUID.randomUUID().toString();
      
 
        CorrelationDataExt correlationData = new CorrelationDataExt();
        correlationData.setId(msgId);
        correlationData.setData("xxxx");
 
 
        rabbitTemplate.convertAndSend(TopicRabbitMqConfig.exchange, "xxx", "message", correlationData);
    }
 
}

3.2 CorrelationDataExt ，扩展CorrelationData ，方便把我们发送出的消息挂回到生产者确认的回调里，这部分的使用方式不确定是否一定对，因为我也是第一次用，但是发现CorrelationData 里只有id能拿到，却拿不到数据，后来看了一个帖子可以扩展CorrelationData ，能实现我想要的效果，还希望路过的大神能指导一下，生产者确认的回调里，如果失败了怎么处理消息？感谢！！

/**
 * CorrelationData的自定义实现，用于拿到消息内容
 * @author coco
 * @date 2022/9/16
 */
public class CorrelationDataExt extends CorrelationData {
    //数据
    private volatile Object data;
 
    public Object getData() {
        return data;
    }
 
    public void setData(Object data) {
        this.data = data;
    }
 
}