当前位置： 首页 > article >正文

RabbitMQ（补档）

article 2025/3/17 12:40:22

        RabbitMQ 是一个开源的消息队列软件（有时也被称为消息代理），它实现了高级消息队列协议（AMQP）。它主要用于应用程序之间，或者软件组件之间的消息通信。通过使用 RabbitMQ，可以实现异步的、可靠的、可扩展的消息处理机制，这使得它成为构建分布式系统时的一个理想选择。

RabbitMQ 应用场景：

        解耦服务：在微服务架构或分布式系统中，服务之间可以通过消息队列进行通信，从而实现松散耦合。这样做的好处是，即使某些服务出现故障，也不会影响到整个系统的运行。

        异步处理：对于一些不需要立即响应的操作，可以采用异步的方式进行处理。例如，用户注册成功后发送欢迎邮件，这类操作可以放到消息队列中异步执行，从而提高用户体验和系统的吞吐量。

        负载均衡：通过将任务分发给多个消费者，RabbitMQ 可以帮助实现负载均衡，确保没有单个节点过载。这特别适用于需要处理大量请求的场景，如高并发的Web应用。

        批量处理：对于需要批量处理的任务，可以先将任务存储在 RabbitMQ 中，等到积累了一定数量的任务后再统一处理。这种方法可以减少数据库的访问次数，提升性能。

RabbitMQ组件:

        生产者(Producer)：发送消息的应用程序。

        消费者(Consumer)：接收消息的应用程序。

        队列(Queue)：存储消息的地方，生产者将消息发送到队列中，消费者从队列中读取消息。

        交换机(Exchange)：负责接收来自生产者的消息，并根据一定的规则将消息路由到相应的队列中。

        绑定(Binding)：定义了交换机和队列之间的关系，即交换机如何将消息分发到具体的队列中。

RabbitMQ的工作流程

1.Producer ⽣产了⼀条消息

2.Producer 连接到RabbitMQBroker, 建⽴⼀个连接(Connection),开启⼀个信道(Channel)

3.Producer 声明⼀个交换机(Exchange), 路由消息

4.Producer 声明⼀个队列(Queue), 存放信息

5.Producer 发送消息⾄RabbitMQ Broker

6.RabbitMQ Broker 接收消息, 并存⼊相应的队列(Queue)中, 如果未找到相应的队列, 则根据⽣产者的配置, 选择丢弃或者退回给⽣产者

RabbitMQ 交换机类型

        RabbitMQ 提供了多种类型的交换机（Exchange），每种类型决定了消息如何从生产者路由到一个或多个队列。以下是 RabbitMQ 中主要的交换机类型：

1. 直连交换机（Direct Exchange）
        功能：直连交换机根据消息携带的路由键（Routing Key）和绑定键（Binding Key）是否完全匹配来决定将消息路由到哪个队列。
        使用场景：当需要基于不同条件将消息分发给不同的消费者时非常有用。例如，日志系统中可以根据严重性级别（如 info、warning、error）将消息路由到不同的目的地。

2. 主题交换机（Topic Exchange）
        功能：主题交换机允许使用通配符进行模式匹配路由键来路由消息。其中 * 匹配一个单词，# 匹配零个或多个单词。
        使用场景：适合于复杂的消息路由逻辑，尤其是当需要灵活地路由消息到多个消费者时。比如，路由所有来自美国或者欧洲的订单信息，可以使用类似 *.us.* 或 *.eu.* 的模式。

3. 扇形交换机（Fanout Exchange）
        功能：扇形交换机忽略路由键，直接将消息广播到所有绑定到该交换机的队列上。
        使用场景：当你想向多个消费者同时发送相同的消息时非常适合。例如，实时更新或广播通知系统。

4. 头信息交换机（Headers Exchange）
        功能：头信息交换机允许你匹配消息头部属性而不是路由键来进行消息路由。它提供了更大的灵活性，因为可以设置复杂的匹配规则，包括AND和OR操作。
        使用场景：适用于需要更复杂过滤条件的消息路由，特别是在需要考虑多于一个参数的情况下。

5. 默认交换机
        虽然严格意义上不被视为一种独立类型，但 RabbitMQ 提供了一个默认的无名交换机（也称为“匿名”交换机）。当你在声明队列时没有指定交换机，消息会通过这个默认交换机直接路由到指定名称的队列。这实际上是一个直连交换机，它的特殊之处在于它是隐式的，且每个队列都自动绑定到它，路由键就是队列的名字。

RabbitMQ使用

1.安装RabbitMQ：RabbitMQ: One broker to queue them all | RabbitMQ
1.引入依赖
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
2.配置文件（生产者和消费者服务都要配置）
server:
  port: 9301
spring:
  application:
    name: test-service
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
3.配置交换机以及绑定队列

        在 RabbitMQConfig 中，我们通过 @Bean 方法定义并返回交换机（Exchange）、队列（Queue）以及绑定（Binding）。这些配置会被 Spring 自动注册到 RabbitMQ 中。

  • 交换机名：在配置中定义的交换机名（return new XxxExchange("交换机名")）会被 Spring 注册到 RabbitMQ 中，决定了消息指定发送到配置的哪一个交换机上。
  • 队列名：队列名（如 direct.queue、topic.queue 等）也会被注册到 RabbitMQ 中，在后续消费监听中被用于指定监听的队列。
  • 绑定：通过 BindingBuilder 将队列绑定到交换机，并指定路由键（Routing Key）或匹配规则。

        这些配置的目的是在 RabbitMQ 中创建好交换机和队列，并建立它们之间的关系。

@Configuration
public class RabbitMQConfigure {
    // Direct Exchange
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("direct.exchange");
    }
    @Bean
    public Queue directQueue() {
        return new Queue("direct.queue");
    }
    @Bean
    public Binding directBinding() {
        return BindingBuilder.bind(directQueue()).to(directExchange()).with("direct.routingKey");
    }

    // Topic Exchange
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("topic.exchange");
    }
    @Bean
    public Queue topicQueue() {
        return new Queue("topic.queue");
    }
    @Bean
    public Binding topicBinding() {
        return BindingBuilder.bind(topicQueue()).to(topicExchange()).with("topic.*");
    }

    // Fanout Exchange
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanout.exchange");
    }
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout.queue1");
    }
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout.queue2");
    }
    @Bean
    public Binding fanoutBinding1() {
        return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
    }
    @Bean
    public Binding fanoutBinding2() {
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    }

    // Headers Exchange
    @Bean
    public HeadersExchange headersExchange() {
        return new HeadersExchange("headers.exchange");
    }
    @Bean
    public Queue headersQueue() {
        return new Queue("headers.queue");
    }
    @Bean
    public Binding headersBinding() {
        return BindingBuilder.bind(headersQueue()).to(headersExchange()).where("headerKey").matches("headerValue");
    }
}
4.发送消息

        在发送消息时，我们使用 RabbitTemplate 的 convertAndSend 方法，并通过第一个参数指定交换机的名称。

public void convertAndSend(String exchange, String routingKey, Object object)

  • exchange: 指定消息发送到的交换机名称。
  • routingKey: 指定消息的路由键，用于决定消息如何路由到队列。
  • message: 要发送的消息内容。

        发送消息时，交换机的名称必须与配置中定义的交换机名一致。

        路由键的使用方式取决于交换机的类型（如 Direct、Topic、Fanout 等），FanoutExchange不注重路由键，不用设置。而TopicExchange和DirectExchange的区别在于路由键支持模糊匹配，而DirectExchange是直接指定路由键。

@Service
public class MessageSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendDirectMessage(String message) {
        rabbitTemplate.convertAndSend("direct.exchange", "direct.routingKey", message);
    }

    public void sendTopicMessage(String message) {
        rabbitTemplate.convertAndSend("topic.exchange", "topic.routingKey", message);
    }

    public void sendFanoutMessage(String message) {
        rabbitTemplate.convertAndSend("fanout.exchange", "", message);
    }

    public void sendHeadersMessage(String message) {
        MessageProperties properties = new MessageProperties();
        properties.setHeader("headerKey", "headerValue");
        Message msg = new Message(message.getBytes(), properties);
        rabbitTemplate.convertAndSend("headers.exchange", "", msg);
    }
}
5.接收消息

        @RabbitListener(queues = "队列名")

        我们使用 @RabbitListener 注解，指定队列名来监听队列收到的消息，而不是交换机，当消息被路由到该队列时，监听方法会自动触发并处理消息。

@Service
public class MessageReceiver {

    @RabbitListener(queues = "direct.queue")
    public void receiveDirectMessage(String message) {
        System.out.println("Received direct message: " + message);
    }

    @RabbitListener(queues = "topic.queue")
    public void receiveTopicMessage(String message) {
        System.out.println("Received topic message: " + message);
    }

    @RabbitListener(queues = "fanout.queue1")
    public void receiveFanoutMessage1(String message) {
        System.out.println("Received fanout message 1: " + message);
    }

    @RabbitListener(queues = "fanout.queue2")
    public void receiveFanoutMessage2(String message) {
        System.out.println("Received fanout message 2: " + message);
    }

    @RabbitListener(queues = "headers.queue")
    public void receiveHeadersMessage(String message) {
        System.out.println("Received headers message: " + message);
    }
}

6.使用流程总结

        Config 配置：定义交换机、队列和绑定关系，这些配置会被注册到 RabbitMQ 中。

        发送消息：通过 RabbitTemplate 指定交换机名称和路由键，将消息发送到交换机。

        监听消息：通过 @RabbitListener 监听队列。


http://www.kler.cn/a/588354.html

相关文章：

  • Hive SQL 精进系列：解锁 Hive SQL 中 KeyValue 函数的强大功能
  • 微信小程序刷题逻辑实现：技术揭秘与实践分享
  • sensor数据在整个rk平台的框架流程是怎么样，
  • 业务幂等性设计的六种方案
  • 蓝桥杯[阶段总结] 二分，前缀和
  • 华为云容器引擎应用场景
  • 游戏成瘾与学习动力激发研究——多巴胺脉冲式释放与奖赏预测误差机制的神经科学解析
  • ccf3501密码
  • 计算机操作系统进程（4）
  • 【网络】什么是反向代理Reverse Proxies？
  • matlab中如何集成使用python
  • Python中在类中创建对象
  • 基于Spring Boot的航司互售系统
  • Java中队列（Queue）和列表（List）的区别
  • 基于ssm+vue汽车租赁系统
  • 量化交易学习笔记02：双均线策略
  • java项目之基于ssm的药店药品信息管理系统（源码+文档）
  • TCP/IP协议中三次握手（Three-way Handshake）与四次挥手（Four-way Wave）
  • 面试系列｜蚂蚁金服技术面【1】
  • C++那些事儿：访问控制与友元函数的奇妙冒险