rabbitmq的高级特性

news2025/10/27 21:58:57

1.消息可靠性问题（实现消费者确认）

1.在项目中添加配置

logging:
  pattern:
    dateformat: HH:mm:ss:SSS
  level:
    cn.itcast: debug
spring:
  rabbitmq:
    host: 192.168.2.182 # rabbitMQ的ip地址
    port: 5672 # 端口
    username: root
    password: root
    virtual-host: /
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true

2.编写returnCallback

3.编写ComfirmCallback

2.消息持久化

可视化界面创建队列的可持久化

代码中的持久化（默认即为持久化）

@Configuration
public class CommonConfig {
    @Bean//创建交换机
    public DirectExchange simpleDirect(){
        return new DirectExchange("simple.direct",true,false);
    }
    @Bean //创建队列
    public Queue simoleQueue(){
        return QueueBuilder.durable("simple.queue").build();
    }
}

@Test
    public void testDurableMessage(){
        //1.准备消息
        Message message = MessageBuilder.withBody("hello,spring".getBytes(StandardCharsets.UTF_8))
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
        //发送消息
        rabbitTemplate.convertAndSend("simple.queue",message);
    }

消费者消息确认机制

消费者失败重试

实现代码

面试常问：

死信交换机

TTL实现流程

延迟队列

安装DelayExchange插件（基于centos7）

官方下载地址：Community Plugins — RabbitMQ

使用命令查看rabbitmq的挂载地址

docker volume inspect mq-plugins

将下载的插件放在插件所在位置

进入容器内部

docker exec -it mq bash
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

基本使用

延迟时间由发送信息的时候设定

使用java代码实现创建延迟队列

创建交换机

发送消息

防止延迟队列被当成错误解决

  @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        //获取rabbitTemplate
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
        //配置returncallback
        rabbitTemplate.setReturnCallback((message, replayCode, replyText, exchange, routingKey) -> {
            if (message.getMessageProperties().getReceivedDelay()>0) {
                return;
            }
            //记录日志
            log.error(
                    "消息发送到队列失败，响应码：{}，失败原因：{}，交换机：{}，路由key：{}，消息：{}",
                    replayCode, replyText, exchange, routingKey,message.toString());
        });
    }
}