目录

RabbitMQ对应的架构

RabbitMQ快速入门

数据隔离

SpringAMQP

消息发送

消息监听

WorkQueues模型

Fanout交换机

Direct交换机

Topic交换机

声明队列和交换机

基本API

fanout实例

direct示例

基于注解声明

消息转换器

RabbitMQ对应的架构

publisher：生产者，也就是发送消息的一方

consumer：消费者，也就是消费消息的一方

queue：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待消费者处理

exchange：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。virtual host：虚拟主机，起到数据隔离的作用。每个虚拟主机相互独立，有各自的exchange、queue

由图可知过程是生产者发消息->交换机接收并路由给队列->消费者

RabbitMQ快速入门

完成一个连接交换机与队列并发送消息的demo

首先打开rabbitmq的浏览器管理端，进入创建队列，选择Classic(经典队列)，再创建队列

之后从All queues便可以看到所创建的队列

来到exchange交换机界面，选择amp.fanout交换机

让交换机与队列绑定，也就是路由

已经绑定的队列能勾在表格看到

然后在publish message中发送消息

之后就可以在交换机的消息速率监控图看到交换速率

黄色线：Publish (In)代表「消息入站速率」，也就是生产者往这个交换机发消息的速度（单位：条 / 秒）。

蓝色线：Publish (Out)代表「消息出站速率」，也就是交换机把消息转发到绑定队列的速度（单位：条 / 秒）。

之后在queues界面可以点击对应的队列查看对应的消息

数据隔离

实操一下数据隔离的操作

在Admin界面可以看到当前显示的虚拟主机，登录的用户，能够访问的虚拟主机，这里我们默认的guest用户能访问名叫“/”的虚拟主机并操作

现在新建用户

tags是对应的权限管理，这里选择Admin即最高权限

这里看到最开始是没有绑定对应的虚拟主机，显示没有能够访问的虚拟主机

现在登录新建的账户来创建虚拟主机

现在即可看到hmall绑定了对应的虚拟主机

这里就可以看到不同的交换机

同时可以看到在新创建的/hmall主机没有在/创建的队列数据，实现了数据隔离

SpringAMQP

将来我们开发业务功能的时候，肯定不会在控制台收发消息，而是应该基于编程的方式。由于RabbitMQ采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。但是，RabbitMQ官方提供的Java客户端编码相对复杂，一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于SpringBoot对其实现了自动装配，使用起来非常方便。

SpringAmqp的官方地址：

Spring AMQP

  • 自动声明队列、交换机及其绑定关系

  • 基于注解的监听器模式，异步接收消息

  • 封装了RabbitTemplate工具，用于发送消息

基于这样一个简单模型来进行一个入门案例

消息发送

步骤如下：引入Spring AMQP->配置连接路径->创建RabbitTemplate->调用convertAndSend()方法

父工程直接引入Spring AMQP

配置rabbitmq的信息

创建RabbitTemplate，并调用方法发送消息

消息监听

前面配置与上面同理，最后引用一个@RabbitLisener的注解，并标注队列名称

WorkQueues模型

思想是让一个queue绑定多个consumer，让处理任务速度加快

创建一个work.queue

然后publish发送50条消息模拟消息堆积情况

建立两个消费者，发现两个消费者的接收消息是一人一条，即轮询

我们将两个消费者指定sleep时间模拟两个消费者的处理效率

发现到最后仍然是每人25条分配

为了实现能者多劳，我们需要在yml里添加一个参数，意义是处理完当前消息再接收其他消息进行处理

最终发现消费者1的效率高并且处理的多，达成了我们的目的

Fanout交换机

Fanout在MQ中叫广播。 在广播模式下，消息发送流程是这样的：

  • 1）  可以有多个队列

  • 2）  每个队列都要绑定到Exchange（交换机）

  • 3）  生产者发送的消息，只能发送到交换机

  • 4）  交换机把消息发送给绑定过的所有队列

  • 5）  订阅队列的消费者都能拿到消息

注意，发送给交换机消息需要改变方法参数

@Test
public void testFanoutExchange() {
    // 交换机名称
    String exchangeName = "hmall.fanout";
    // 消息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

交换机的作用是什么？

  • 接收publisher发送的消息

  • 将消息按照规则路由到与之绑定的队列

  • 不能缓存消息，路由失败，消息丢失

  • FanoutExchange的会将消息路由到每个绑定的队列

Direct交换机

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

在Direct模型下：

  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）

  • 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。

  • Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

需要在交换机绑定队列的时候配置Routing key

发送信息时配置Routing key就可以发送到配置有指定key的队列

@Test
public void testSendDirectExchange() {
    // 交换机名称
    String exchangeName = "hmall.direct";
    // 消息
    String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

描述下Direct交换机与Fanout交换机的差异？

  • Fanout交换机将消息路由给每一个与之绑定的队列

  • Direct交换机根据RoutingKey判断路由给哪个队列

  • 如果多个队列具有相同的RoutingKey，则与Fanout功能类似

Topic交换机

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定BindingKey 的时候使用通配符！

BindingKey 一般都是有一个或多个单词组成，多个单词之间以.分割，例如： item.insert

通配符规则：

  • #：匹配一个或多个词

  • *：匹配不多不少恰好1个词

举例：

  • item.#：能够匹配item.spu.insert 或者 item.spu

  • item.*：只能匹配item.spu

假如此时publisher发送的消息使用的RoutingKey共有四种：

  • china.news代表有中国的新闻消息；

  • china.weather 代表中国的天气消息；

  • japan.news 则代表日本新闻

  • japan.weather 代表日本的天气消息；

解释：

  • topic.queue1：绑定的是china.# ，凡是以 china.开头的routing key 都会被匹配到，包括：

    • china.news

    • china.weather

  • topic.queue2：绑定的是#.news ，凡是以 .news结尾的 routing key 都会被匹配。包括:

    • china.news

    • japan.news

描述下Direct交换机与Topic交换机的差异？

  • Topic交换机接收的消息RoutingKey必须是多个单词，以 **.** 分割

  • Topic交换机与队列绑定时的bindingKey可以指定通配符

  • #：代表0个或多个词

  • *：代表1个词

声明队列和交换机

在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时，队列和交换机是程序员定义的，将来项目上线，又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来，交给运维。在这个过程中是很容易出现错误的。因此推荐的做法是由程序启动时检查队列和交换机是否存在，如果不存在自动创建。

基本API

SpringAMQP提供了一个Queue类，用来创建队列：

Spring AMQP提供了Exchange接口可以来表示不同的交换机

fanout实例

package com.itheima.consumer.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfig {
    /**
     * 声明交换机
     * @return Fanout类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("hmall.fanout");
    }

    /**
     * 第1个队列
     */
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

direct示例

因为需要Routing Key所以在返回Binding时需要用with()带上Routing Key

package com.itheima.consumer.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DirectConfig {

    /**
     * 声明交换机
     * @return Direct类型交换机
     */
    @Bean
    public DirectExchange directExchange(){
        return ExchangeBuilder.directExchange("hmall.direct").build();
    }

    /**
     * 第1个队列
     */
    @Bean
    public Queue directQueue1(){
        return new Queue("direct.queue1");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
    }
    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue directQueue2(){
        return new Queue("direct.queue2");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
    }
    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
    }
}

基于注解声明

基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue1"),
    exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
    System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue2"),
    exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
    System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
}

消息转换器

Spring的消息发送代码接收的消息体是一个Object：

而在数据传输时，它会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

  • 数据体积过大

  • 有安全漏洞

  • 可读性差

我们需要引用jackjon依赖，来序列化

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>

注意，如果项目中引入了spring-boot-starter-web依赖，则无需再次引入Jackson依赖。

我们在consumer服务中定义一个新的消费者，publisher是用Map发送，那么消费者也一定要用Map接收，格式如下：

@RabbitListener(queues = "object.queue")
public void listenSimpleQueueMessage(Map<String, Object> msg) throws InterruptedException {
    System.out.println("消费者接收到object.queue消息：【" + msg + "】");
}

# rabbitmq # 分布式 # java # 开发语言
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

当 AI 修图师装上“万能插头“：MCP 如何改变修图工作流

avatar AI编程社区
cover

2026年8月开发者实战指南：ChatGPT Plus / Pro + Codex 如何协同完成需求分析、编码、测试、调试与代码审查

avatar AI编程社区
cover

Claude API + Python实战：结构化输出后，如何拦住“看起来正确”的JSON

avatar AI编程社区