rabbitmq的confirm模式获取correlationData为null解决办法

news2024/11/26 22:36:13

回调函数confirm中的correlationData=null

// 实现confirm回调,发送到和没发送到exchange,都触发
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    // 参数说明:
    // correlationData: 相关数据,可以在发送消息时,进行设置该参数
    // ack: 结果
    // cause: 原因

    if (ack) {
        log.info("【ConfirmCallback】消息已经送达Exchange，ack已发");

    } else {
        ReturnedMessage message = correlationData.getReturned();

        if (message != null) {
            String msgData = new String(message.getMessage().getBody());
            log.error("消息发送到 exchange {} 失败,原因: {},id: {}, routingKey: {},body: {}", message.getExchange(), cause, correlationData.getId(), message.getRoutingKey(), msgData);
        } else {
            log.error("消息发送 exchange 失败,原因: {},id: {}", correlationData.getId(),cause);
        }

    }
}

解决办法

在convertAndSend方法中传入correlationData数据

@SpringBootTest
class RabbitmqDemoApplicationTests {

    @Test
    void contextLoads() {
    	// 模拟消息
        BattleSubmitMqVo msg = new BattleSubmitMqVo().setUserId(1L).setRoomId("123").setTimes(300L);
        // 工具类发送消息到mq
        MqUtil.sendMsgToMq(RabbitConfig.BATTLE_PAPER_EXCHANGE,RabbitConfig.BATTLE_PAPER_ROUTING_KEY, msg);

    }

}

工具类

package com.example.rabbitmqdemo.util;

import cn.hutool.json.JSONUtil;
import com.sun.istack.internal.NotNull;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * desc:
 *
 * @author qts
 * @date 2023/11/3 0003
 */
@Component
public class MqUtil {

    private static RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate2;

    @PostConstruct
    public void init(){
        rabbitTemplate = rabbitTemplate2;
    }

    /**
     * 发送消息并
     * 添加 CorrelationData数据
     * @param exchange
     * @param routingKey
     * @param msg
     */
    public static void sendMsgToMq(String exchange, String routingKey, Object msg){
        CorrelationData correlationData = new CorrelationData();
        correlationData.setReturned(new ReturnedMessage(new Message(JSONUtil.toJsonStr(msg).getBytes()),1,"1",exchange,routingKey));

        rabbitTemplate.convertAndSend(exchange,routingKey,msg,correlationData);
    }

    /**
     * 发送消息
     * 添加 CorrelationData数据, 消息后处理回调
     * @param exchange
     * @param routingKey
     * @param msg
     * @param messagePostProcessor 消息后处理回调
     */
    public static void sendMsgToMq(String exchange, String routingKey, Object msg,MessagePostProcessor messagePostProcessor){
        CorrelationData correlationData = new CorrelationData();
        correlationData.setReturned(new ReturnedMessage(new Message(JSONUtil.toJsonStr(msg).getBytes()),1,"1",exchange,routingKey));

        rabbitTemplate.convertAndSend(exchange,routingKey,msg,messagePostProcessor,correlationData);
    }
}