RabbitMQ个人理解与基本使用

news2025/7/19 12:44:01

一. 作用：

二. RabbitMQ的5中队列模式：

1. 简单模式

2. Work模式

3. 发布/订阅模式

4. 路由模式

5. 主题模式

三. 消息持久化：

消息过期时间

ACK应答

四. 同步接收和异步接收：

应用场景

五. 基本使用：

引入依赖库：

配置文件RabbitMQConfig：

创建消息任务类：

解析：

一. 作用：

RabbitMQ主要用于消息队列的实现。

二. RabbitMQ的5中队列模式：

1. 简单模式

一个生产者（发送方）对应一个消费者（接收方）

2. Work模式

一个生产者对应多个消费者，但是只能有一个消费者获得消息（排他）

3. 发布/订阅模式

一个消费者将消息首先发送到fanout交换器，交换器绑定到多个队列，然后与之对应的所有消费者都能接收到消息（不排他）

4. 路由模式

生产者将消息发送到direct交换器，交换器按照关键字（Key），把消息路由到某个队列

5. 主题模式

生产者将消息发送到Topic交换器，交换器按照复杂的规则，把消息路由到某个队列

三. 消息持久化：

消息的可靠性是RabbitMQ的一大特色，那么RabbitMQ是如何保证消息可靠性的呢？答案就是消息持久化。持久化可以防止在异常情况下丢失数据。除了消息持久化之外，甚至交换器和队列都能持久化。也就是说rabbitmq的消息会被存储在磁盘上，只有当消费收到消息，rabbitmq确认消费者收到消息（Acknowledgments--简称ACK）后才会将消息从队列中删除。

消息过期时间

如果消费者一直不接收消息，消息会一直保存在消息队列当中，短期内可能不会有什么影响，但是如果经过长时间的积累后，消息会变得很多很多，浪费大量的资源，内存。

为了应对这种情况，就可以对rabbitmq设置消息的过期时间，在规定时间内消息没有被接收，就会删除掉该消息。

ACK应答

消费者接收到消息后，为了让RabbitMQ 知道，就需要返回一个ACK应答，告诉RabbitMQ消费者已经收到了消息，如果收到消息后我们需要删除该消息，只需要在ACK应答中加上deliveryTag标志位。

四. 同步接收和异步接收：

同步接收：指消费者调用方法时，会阻塞来等待消息，直到消息被成功消费或者队列为空。（没有消息等待消息再接着处理）。

异步接收：指消费者不会在接收消息时阻塞，而是通过回调函数处理消息。消费者在等待消息的同时不会停下，可以处理其他任务。（当有消息时才来处理消息）。

应用场景

同步接收：当消息的处理顺序对业务逻辑非常重要，就使用同步接收，消费者一次只处理一个消息，确保了每条消息的处理顺序。

异步接收：当处理消息的时间比较长，或者系统的并发量大时，采用异步接收会更好。

RabbitMQ还有一个杀手锏——同时使用异步收发和同步收发。

五. 基本使用：

引入依赖库：

<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.9.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置文件RabbitMQConfig：

import com.rabbitmq.client.ConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    @Value("${rabbitmq.factoryHost}")
    private String host;

    @Bean
    public ConnectionFactory connectionFactory() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setPort(5672);
        return factory;
    }
}

host配置，我将rabbitMQ放在虚拟机上的，所有ip是虚拟机的地址：

创建消息任务类：

@Slf4j
@Component
public class MessageTask {
    @Autowired
    private ConnectionFactory factory;
    @Autowired
    private MessageService messageService;

    /*
    * 同步发送消息
    * */
    public void send(String topic, MessageEntity entity) {
        //向MongoDB保存消息数据，返回消息ID
        String id = messageService.insertMessage(entity);
        //向RabbitMQ发送消息
        try(Connection connection = factory.newConnection();
            Channel channel = connection.createChannel()){
            //连接到某个topic
            channel.queueDeclare(topic, true, false, false, null);
            HashMap header = new HashMap();
            header.put("messageId",id);
            //创建AMQP协议参与对象，添加附加属性
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().headers(header).build();
            channel.basicPublish("",topic,properties,entity.getMsg().getBytes());
            log.debug("消息发送成功");
        } catch (Exception e){
            log.error(e.getMessage());
            throw new EmosException("向MQ发送消息失败");
        }
    }

    /*
    * 异步发送消息
    * */
    @Async("AsyncTaskExecutor")
    public void sendAsync(String topic, MessageEntity entity) {
        send(topic, entity);
    }

    /*
    * 同步接收消息
    * */
    public int receive(String topic) {
        int i = 0;
        try (//接收消息数据
             Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 从队列中获取消息，不自动确认
            channel.queueDeclare(topic, true, false, false, null);
            //Topic中有多少条数据未知，所以使用死循环接收数据，直到接收不到消息，退出死循环
            while (true) {
                //创建响应接收数据，禁止自动发送Ack应答
                GetResponse response = channel.basicGet(topic, false);
                if (response != null) {
                    AMQP.BasicProperties properties = response.getProps();
                    Map<String, Object> header = properties.getHeaders(); //获取附加属性对象
                    String messageId = header.get("messageId").toString();
                    byte[] body = response.getBody();//获取消息正文
                    String message = new String(body);
                    log.debug("从RabbitMQ接收的消息：" + message);
                    MessageRefEntity entity = new MessageRefEntity();
                    entity.setMessageId(messageId);
                    entity.setReceiverId(Integer.parseInt(topic));
                    entity.setReadFlag(false);
                    entity.setLastFlag(true);
                    messageService.insertRef(entity); //把消息存储在MongoDB中
                    //数据保存到MongoDB后，才发送Ack应答，让Topic删除这条消息
                    long deliveryTag = response.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag, false);
                    i++;
                } else {
                    break; //接收不到消息，则退出死循环
                }
            }
        } catch (Exception e) {
            log.error("执行异常", e);
        }
        return i;
    }

    /*
    * 异步接收消息
    * */
    @Async
    public int receiveAsync(String topic) {
        return receive(topic);
    }

    /*
    * 同步删除消息
    * */
    public void deleteQueue(String topic) {
        try(//接收消息数据
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel()){
            channel.queueDelete(topic);
            log.debug("成功删除消息队列:"+topic);
        } catch (Exception e){
            log.error("删除消息队列失败:",e);
            throw new EmosException("删除消息队列失败");
        }
    }

    /*
    * 异步删除消息
    * */
    @Async
    public void deleteAsync(String topic) {
        deleteQueue(topic);
    }
}

解析：

channel.queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments);

queueName:队列的名称，用于标识消息的存储位置。
durable：

true，表示队列是持久化的。

false，表示队列是非持久化的。

exclusive：

true：队列仅供当前连接使用，连接断开时队列会自动删除。

false：队列可供多个连接共享。

autoDelete：
true：当队列不再被任何消费者订阅时，队列会自动删除。
false：队列即使没有消费者订阅也会一直存在，直到手动删除。
arguments：额外的参数，null表示没有额外参数

Map<String, Object> arguments = new HashMap<>();
arguments.put("x-message-ttl", 60000); // 设置队列中消息的过期时间为 60 秒（60000 毫秒）

channel.queueDeclare("myQueue", true, false, false, arguments);

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2261281.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！