您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生

尚硅谷RabbitMQ视频-022-进阶篇-可靠性-生端确-测试

发布时间：2026-11-03 00:16:36编辑：雪饮阅读（）

本篇主要是生产者生产消息到交换机和队列的可靠性，通过回调观察是否发送到交换机以及是否到达队列的回调消息。
首先是src\main\resources\application.yml:

spring:

  rabbitmq:

    host: localhost

    port: 5672

    username: guest

    password: 123456

    virtual-host: /

    publisher-confirm-type: correlated #交换机的确认

    publisher-returns: true #队列的确认

logging:

  level:

    org.example.springBoot.listener.MyMessageListener: info

    org.example.confirmProducer.listener.MyMessageListener: info


publisher-confirm-type、publisher-returns是基于上篇我们新增的，然后

org.example.confirmProducer.listener.MyMessageListener: info

这个也是新增的。
接下来是新增RabbitMQConsumerMainType.java：

package org.example.confirmProducer;



import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;



@SpringBootApplication

public class RabbitMQConsumerMainType {

    //运行main方法使得消费者监听队列(注意默认会占用本地8080端口，若已经被占用，则可能无法启动)

    public static void main(String[] args) {

        SpringApplication.run(RabbitMQConsumerMainType.class, args);

    }

}

注意观察，区别于上篇中的包。
后续几个都是新定义的包里面的。一旦包对不上了，就逻辑混乱了，可能就回调不上了。
然后
MyMessageListener.java：

package org.example.confirmProducer.listener;



import com.rabbitmq.client.Channel;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.annotation.Exchange;

import org.springframework.amqp.rabbit.annotation.Queue;

import org.springframework.amqp.rabbit.annotation.QueueBinding;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;



//没有这个注解，消费者不会监听队列，也不会接收消息

@Component

@Slf4j

public class MyMessageListener {

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";

    public static final String ROUTING_KEY = "order";

    public static final String QUEUE_NAME = "queue.order";



    /*

    * 监听队列，不存在则创建

    * */

    @RabbitListener(bindings = @QueueBinding(

            value = @Queue(value = QUEUE_NAME, durable = "true"),

            //这里仅仅指定了交换机名称，没有指定交换机类型，默认是direct交换机类型

            exchange = @Exchange(value = EXCHANGE_DIRECT),

            key = {ROUTING_KEY}

    ))



    // 写法二：仅仅监听队列，不自动创建

//@RabbitListener(queues = {QUEUE_NAME})

    public void processMessage(String dataString, Message message, Channel channel) {

        System.out.println("test confirmProducer listener 消费端接收到了消息：" + dataString);

    }

}

然后RabbitConfig.java：

package org.example.confirmProducer.config;

import jakarta.annotation.PostConstruct;

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.context.annotation.Configuration;

import lombok.extern.slf4j.Slf4j;

@Configuration

@Slf4j

public class RabbitConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    @Autowired

    /*

    这里容易出现这个问题

    无法自动装配。找不到 'RabbitTemplate' 类型的 Bean。

    需要在类级别的注解上增加注解:

    @SpringBootApplication(scanBasePackages = "org.example")

    这里的注解扫描包名要同一个包或者父级包



    这里局部变量 可能报错：无法自动装配。找不到 'RabbitTemplate' 类型的 Bean。，

    这是ide上面的报错，在这里是误报，可以忽略。

    * */

    private RabbitTemplate rabbitTemplate;



    @PostConstruct

    public void initRabbitTemplate() {

        rabbitTemplate.setConfirmCallback(this);

        rabbitTemplate.setReturnsCallback(this);

    }



    @Override

    public void confirm(CorrelationData correlationData, boolean ack, String cause) {

        // 消息发送到交换机成功或失败时调用这个方法

        log.info("confirm() 回调函数打印 CorrelationData: " + correlationData);

        log.info("confirm() 回调函数打印 ack: " + ack);

        log.info("confirm() 回调函数打印 cause: " + cause);

    }



    @Override

    public void returnedMessage(ReturnedMessage returned) {

        // 发送到队列失败时才调用这个方法

        log.info("消息主体: " + new String(returned.getMessage().getBody()));

        log.info("应答码: " + returned.getReplyCode());

        log.info("描述: " + returned.getReplyText());

        log.info("消息使用的交换器 exchange : " + returned.getExchange());

        log.info("消息使用的路由键 routing : " + returned.getRoutingKey());

    }

}

然后RabbitMQTest.java：

package org.example.confirmProducer;



import org.junit.jupiter.api.Test;

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

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;



@SpringBootTest

public class RabbitMQTest {

public static final String EXCHANGE_DIRECT = "exchange.direct.order";

    public static final String ROUTING_KEY = "order";



    @Autowired

    private RabbitTemplate rabbitTemplate;



    @Test

    public void test01SendMessage() {

        //正常发送

        //rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, ROUTING_KEY, "send test confirmProducer");

        //故意改错交换机然后发送

        //rabbitTemplate.convertAndSend(EXCHANGE_DIRECT+"error", ROUTING_KEY, "send test confirmProducer");

        //故意改错队列(让其匹配不到队列)然后发送

        rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, ROUTING_KEY+"error", "send test confirmProducer");

    }

}

最后运行的时候要先运行监听
RabbitMQConsumerMainType.java中的main方法
然后运行RabbitMQTest.java中的test01SendMessage方法