rabbitmq基本操作

集成springboot

新增依赖

       <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

yml配置

spring:
  rabbitmq:
    host: localhost
    username: guest
    password: 123456
    virtual-host: /
    port: 5672
    publisher-confirm-type: CORRELATED
    publisher-returns: true

创建监听器
1. @RabbitListener 注解中配置在项目启动后，若没有该交换机、路由、队列的话，会自动生成
2. processMessage 接收生产者发送的消息

package com.gyq.common.listener;

import com.rabbitmq.client.Channel;
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;

/**
 * @author GYQ
 */
@Component
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"),
            exchange = @Exchange(value = EXCHANGE_DIRECT),
            key = {ROUTING_KEY}
    ))
    public void processMessage(String dataString, Message message, Channel channel) {
        System.out.println("消费端接收到了消息：" + dataString);
    }
}

生产者发送消息

  @Resource
    private RabbitTemplate rabbitTemplate;
    public static final String EXCHANGE_DIRECT = "exchange.direct.order";
    public static final String ROUTING_KEY = "order";
    public static final String QUEUE_NAME = "queue.order";

    @Test
    void test() {
        rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, ROUTING_KEY, "我来助你！");
    }

消息丢失怎么办

消息的流程

项目初始化的时候先创建对应的交换机、路由、队列等
生产者生产消息--交换机将消息发送到对应的队列
消费者在队列中获取相应的消息

【1】生产者没有将消息打到队列

回调函数日志追踪

import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @author GYQ
 * @date 2024/9/23 10:28
 */
@Configuration
public class RabbitConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    @Resource
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void initRabbitTemplate() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }

    /**
     * 发送到交换机后，无论成功与否的回调函数
     *
     * @param correlationData
     * @param ack             成功还是失败
     * @param cause           原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("correlationData：" + correlationData);
        System.out.println("ack：" + ack);
        System.out.println("cause：" + cause);
    }

    /**
     * 交换机发送到队列后失败才会调用该方法
     *
     */
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        System.out.println("消息主题：" + new String(returnedMessage.getMessage().getBody()));
        System.out.println("应答码：" + returnedMessage.getReplyCode());
        System.out.println("描述：" + returnedMessage.getReplyText());
        System.out.println("交换器：" + returnedMessage.getExchange());
        System.out.println("路由：" + returnedMessage.getRoutingKey());
    }
}

添加备用交换机，队列（我没这么用过，感觉没必要，具体看业务）

【2】消息队列服务器宕机

消息是持久化到硬盘的，重启就ok

【3】消费端宕机或者抛异常导致消息没有被消费

获取返回信息，做是否重新放入队列处理

import com.rabbitmq.client.Channel;
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;

import java.io.IOException;

/**
 * @author GYQ
 */
@Component
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";

    /**
     * 对消息处理，若消息接收失败，重发
     * 还是失败，销毁消息
     *
     * @param dataString 消息
     * @param message
     * @param channel
     */

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = QUEUE_NAME, durable = "true"),
            exchange = @Exchange(value = EXCHANGE_DIRECT),
            key = {ROUTING_KEY}
    ))
    public void processMessage(String dataString, Message message, Channel channel) throws IOException {
        //获取消息唯一id
        long deliveryTag = message.getMessageProperties().getDeliveryTag();


        try {
            //核心处理消息
            System.out.println("消费端接收到了消息：" + dataString);
            int i = 10 / 0;
            //false：表示该条消息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            //获取消息是否为重新投递
            Boolean redelivered = message.getMessageProperties().getRedelivered();
            //第一次捕获异常，返还给队列
            if (!redelivered) {
                channel.basicNack(deliveryTag, false, true);
            } else {
                channel.basicNack(deliveryTag, false, false);
            }
        }

    }
}

代码解析

channel.basicAck(deliveryTag, false); 中的第二个参数 false 指的是 是否进行批量确认（batch acknowledgment）。

在 RabbitMQ 中，basicAck 方法用于确认消息已经被成功消费，并将其从队列中删除。这个方法有两个参数：

deliveryTag：用于标识这条消息的唯一标识符（ID），RabbitMQ 在向消费者传递消息时，会为每条消息分配一个 deliveryTag。
multiple：这个布尔值参数决定了是否批量确认消息：false表示只确认当前这条消息（由 deliveryTag 标识的消息）。 true表示批量确认小于或等于当前 deliveryTag 的所有消息。所以，channel.basicAck(deliveryTag, false); 中的 false 表示只确认这条单独的消息。如果你设置为 true，则会批量确认所有 deliveryTag 小于或等于给定值的消息。

channel.basicNack(deliveryTag, false, true); 用于拒绝消息，且可以选择是否重新入队。它有三个参数：

deliveryTag：与 basicAck 相同，表示要拒绝的消息的唯一标识符（ID）。
multiple：与 basicAck 中的含义相同，表示是否批量处理消息：false*仅拒绝当前 deliveryTag 标识的这条消息。*true批量拒绝所有 deliveryTag 小于或等于指定值的消息。
requeue：这是一个布尔值，决定被拒绝的消息是否重新放回队列：false将消息直接丢弃，或者根据配置进入死信队列（DLX，Dead Letter Exchange）。所以，channel.basicNack(deliveryTag, false, true); 的含义是：
- 只拒绝当前这条消息，不批量操作；
- 并且将这条消息重新放回队列以供再次消费。

削峰填谷

举个例子，队列中有100条数据，客户端一次能处理1条数据，大量数据从队列获取，客户端maybe会炸，所以可以限制一次从队列拿多少条数据

spring:
  rabbitmq:
    host: localhost
    username: guest
    password: 123456
    virtual-host: /
    port: 5672
    publisher-confirm-type: CORRELATED
    publisher-returns: true
    listener:
      simple:
        acknowledge-mode: manual #把消息确认模式改为手动
        prefetch: 1 #关键配置，一次只拿1条【自己可以设置】

客户端测试代码

 @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = QUEUE_NAME, durable = "true"),
            exchange = @Exchange(value = EXCHANGE_DIRECT),
            key = {ROUTING_KEY}
    ))
    public void processMessage(String dataString, Message message, Channel channel) throws IOException, InterruptedException {
        System.out.println(dataString);
        TimeUnit.SECONDS.sleep(1);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        channel.basicAck(deliveryTag,false);
    }

设置消息过期时间

一个队列中的消息长时间不被消费，这条消息大概率没用，所以可以设置过期时间

有两种设置方法，一种是在客户端设置（可以自行百度），另外一种就是代码设置，如下

 @Test
    void test() {
        //发送消息时添加如下代码
        MessagePostProcessor postProcessor = message -> {
            //设置7000毫秒后过期【成了死信】
            message.getMessageProperties().setExpiration("7000");
            return message;
        };
        for (int i = 0; i < 100; i++) {
            rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, ROUTING_KEY, "我来助你！" + i,postProcessor);
        }

    }

**ps：**如果既在客户端又在代码中都设置了过期时间，那么那个时间小，就默认是哪一个

死信

生成原因

服务端没有接收到消息（nack）
消息超时
消息溢出（超出消息队列的容量）

处理

如果消息不重要，没了就没了，消息重要，新增一个死信队列，将失败的消息转入死信队列，后续可以业务处理（告警，日志等）

主要配置就是在主队列的注解中加一个死信交换机和路由的参数x-dead-letter-exchange x-dead-letter-routing-key

package com.gyq.common.listener;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @author GYQ
 */
@Component
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";


    public static final String DEAD_EXCHANGE_DIRECT = "dead.exchange.direct.order";
    public static final String DEAD_ROUTING_KEY = "dead.order";
    public static final String DEAD_QUEUE_NAME = "dead.queue.order";

  
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = QUEUE_NAME, durable = "true", arguments = {
                    @Argument(name = "x-dead-letter-exchange", value = DEAD_EXCHANGE_DIRECT),
                    @Argument(name = "x-dead-letter-routing-key", value = DEAD_ROUTING_KEY)
            }),
            exchange = @Exchange(value = EXCHANGE_DIRECT),
            key = {
                    ROUTING_KEY
            }
    ))
    public void processMessage(String dataString, Message message, Channel channel) throws IOException, InterruptedException {
        System.out.println(dataString);
        TimeUnit.SECONDS.sleep(1);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        //尝试能否进入死信队列
        channel.basicNack(deliveryTag, false,false);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = DEAD_QUEUE_NAME, durable = "true"),
            exchange = @Exchange(value = DEAD_EXCHANGE_DIRECT),
            key = {DEAD_ROUTING_KEY}
    ))
    public void deadMessage(String dataString, Message message, Channel channel) throws IOException, InterruptedException {
        System.out.println(dataString);
        TimeUnit.SECONDS.sleep(1);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        channel.basicAck(deliveryTag, false);
    }
}

延迟队列

效果：比如买到商品后，提示30min未支付订单无效这种

安装插件

方法，先下载rabbitmq-plugin的rabbitmq-delayed-message-exchange

如果用docker，进入docker中rabbitmq 的容器

//查询rabbitmq的容器
docker ps

//进入容器
docker exec -it 6ce9f06e6d6d /bin/bash

//进入插件目录
 cd plugins

//下载延时插件
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/tag/v3.13.0

//启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

//在启用插件后，可以通过以下命令检查插件是否成功启用
rabbitmq-plugins list

//在列出的插件列表中，启用的插件会有 [E] 标记。应该看到 rabbitmq_delayed_message_exchange 列表中显示为已启用。如下图


//最后重启rabbitmq
rabbitmqctl stop_app

rabbitmqctl start_app

简单例子

生产者生产消息，消费者10秒后接收消息

    /**
     * 生产者发送延时消息
     */

    //延迟队列
    public static final String DELAY_EXCHANGE = "delay.exchange";
    public static final String DELAY_QUEUE = "delay.queue";
    public static final String DELAY_ROUTING_KEY = "delay.routing.key";

    @Test
    void test() {

        MessagePostProcessor postProcessor = message -> {
            //设置10秒后过期
            message.getMessageProperties().setHeader("x-delay", 10000);
            return message;
        };
        LocalDateTime now = LocalDateTime.now();
        String format = now.format(DateTimeFormatter.ofPattern("hh:mm:ss"));
        rabbitTemplate.convertAndSend(DELAY_EXCHANGE, DELAY_ROUTING_KEY, "我来助你！" + format, postProcessor);
    }

//结果
我来助你！10:27:39

    /**
     * 消费者消费延迟消息
     */


    //延迟队列
    public static final String DELAY_EXCHANGE = "delay.exchange";
    public static final String DELAY_QUEUE = "delay.queue";
    public static final String DELAY_ROUTING_KEY = "delay.routing.key";


    // 监听延迟队列的消息
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = DELAY_QUEUE, durable = "true"),
            exchange = @Exchange(value = DELAY_EXCHANGE, type = "x-delayed-message", arguments = {
                    @Argument(name = "x-delayed-type", value = "direct")  // 指定延迟交换机类型
            }),
            key = {DELAY_ROUTING_KEY}
    ))
    public void handleDelayedMessage(String message) {
        System.out.println(LocalDateTime.now().format(DateTimeFormatter.ofPattern("hh:mm:ss")));
        System.out.println("接收到延迟消息: " + message);

    }

//结果
10:27:49
接收到延迟消息: 我来助你！10:27:39

菜单

分享

rabbitmq基本操作

rabbitmq基本操作

集成springboot

消息丢失怎么办

消息的流程

【1】生产者没有将消息打到队列

【2】消息队列服务器宕机

【3】消费端宕机或者抛异常导致消息没有被消费

削峰填谷

设置消息过期时间

死信

生成原因

延迟队列

安装插件

简单例子

评论

【三】springAi聊天记忆功能（本地内存和数据库存储）

juc三种辅助类

基于jdk1.8的jvm总结

为什么 scp 大文件传输会越传越慢，以及我是如何解决的

记录一次webFlux 无法输出bug

【二】springAi实现提问敏感词的过滤(自定义拦截器)

rabbitmq基本操作

【一】SpringAi实现多平台Ai的调用

【五】springAi函数调用

【四】springAi格式化返回功能