SpringAMQP整合RabbitMQ-五种工作模式Demo

Posted 2022-12-03 Henrik-Yao

文章目录

• 一.MQ基本概念
• 二.RabbitMQ
• 三.Docker部署RabbitMQ
• 四.SpringAMQP
• 五.RabbitMQ工作模式
• • 1.Basic Queue 简单模式
  • 2.Work queues 工作队列模式
  • 3.Pub/Sub 订阅模式
  • 4.Routing 路由模式
  • 5.Topics 通配符模式
• 六.消息转换器

一.MQ基本概念

1.概述

• MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器，是存储消息的中间件
• 分布式系统通信有两种方式，一种是直接远程调用（例如Feign），一种是借助类似于rabbitmq的第三方中间件完成间接通信，即异步通讯

2.MQ的优势

• 应用解耦：提高系统容错性和可维护性
• 异步提速：提升用户体验和系统吞吐量
• 削峰填谷：提高系统稳定性

3.几种常见MQ的对比：

	RabbitMQ	ActiveMQ	RocketMQ	Kafka
公司/社区	Rabbit	Apache	阿里	Apache
开发语言	Erlang	Java	Java	Scala&Java
协议支持	AMQP，XMPP，SMTP，STOMP	OpenWire,STOMP，REST,XMPP,AMQP	自定义协议	自定义协议
可用性	高	一般	高	高
单机吞吐量	一般	差	高	非常高
消息延迟	微秒级	毫秒级	毫秒级	毫秒以内
消息可靠性	高	一般	高	一般

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

二.RabbitMQ

1.AMQP

AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等

2.RabbitMQ

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而集群和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

2007年，Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发。
Erlang 语言由 Ericson 设计，专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛

3.RabbitMQ中的相关概念

• Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker
• Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网 络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
• Connection：publisher／consumer 和 broker 之间的 TCP 连接
• Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection 的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的Connection 极大减少了操作系统建立 TCP connection 的开销
• Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到
  queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)
• Queue：消息最终被送到这里等待 consumer 取走
• Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

三.Docker部署RabbitMQ

拉取镜像

docker pull rabbitmq:management

启动docker

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management

四.SpringAMQP

SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便

SpringAMQP提供了三个功能：

• 自动声明队列、交换机及其绑定关系
• 基于注解的监听器模式，异步接收消息
• 封装了RabbitTemplate工具，用于发送消息

五.RabbitMQ工作模式

工作模式	说明
Basic Queue 简单模式	一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）
Work Queue 工作队列模式	一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）
Publish/subscribe 发布订阅模式	需要设置类型为 fanout 的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列
Routing 路由模式	需要设置类型为 direct 的交换机，交换机和队列进行绑定，并且指定 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列
Topic 通配符模式	需要设置类型为 topic 的交换机，交换机和队列进行绑定，并且指定通配符方式的 routing key，当发送消息到交换机后，交换机会根据 routing key 将消息发送到对应的队列

1.Basic Queue 简单模式

• publisher：消息生产者，也就是要发送消息的程序
• consumer：消息的消费者，会一直等待消息到来
• queue：消息队列，类似一个邮箱，可以缓存消息；发布者向其中投递消息，消费者从其中取出消息

代码实现

publisher服务端

①引入依赖

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

②配置MQ地址，在publisher服务的application.yml中添加配置

spring:
  rabbitmq:
    host: 101.43.16.42 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: henrik # 用户名
    password: yao # 密码

③在publisher服务中编写测试类SpringAmqpTest，利用RabbitTemplate实现消息发送

@SpringBootTest
public class SpringAmqpTest 

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSimpleQueue() 
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, spring amqp!";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message);
    

consumer服务端

①引入依赖

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

②配置MQ地址，在consumer服务的application.yml中添加配置

spring:
  rabbitmq:
    host: 101.43.16.42 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: henrik # 用户名
    password: yao # 密码

③consumer服务的listener包中新建一个类SpringRabbitListener

@Component
public class SpringRabbitListener 

    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException 
        System.out.println("spring 消费者接收到消息：【" + msg + "】");
    

④启动consumer服务，然后在publisher服务中运行测试代码，发送MQ消息

---

所有操作均需引入依赖和配置rabbitmq，后续demo不再重复

---

2.Work queues 工作队列模式

• Work Queues：与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息
• 应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度
• 在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系
• Work Queues 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如：短信服务部署多个，只需要有一个节点成功发送即可

publisher服务端

publisher服务中的SpringAmqpTest类中添加一个测试方法

/**
     * workQueue
     * 向队列中不停发送消息，模拟消息堆积。
     */
@Test
public void testWorkQueue() throws InterruptedException 
    // 队列名称
    String queueName = "simple.queue";
    // 消息
    String message = "hello, message_";
    for (int i = 0; i < 50; i++) 
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    

consumer服务端

为了模拟多个消费者绑定同一个队列，在consumer服务的SpringRabbitListener中添加2个新的方法

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException 
    System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(20);


@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException 
    System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);

3.Pub/Sub 订阅模式

在订阅模型中，多了一个 Exchange 角色，而且过程略有变化：

• Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
• Consumer：消费者，消息的接收者，会一直等待消息到来
• Queue：消息队列，接收消息、缓存消息
• Exchange：交换机（X）。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、
  递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：

三种类型的交换机

1. Fanout：广播，将消息交给所有绑定到交换机的队列
2. Direct：定向，把消息交给符合指定routing key 的队列
3. Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列 Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合
   路由规则的队列，那么消息会丢失

consumer端

在springboot添加配置类，借助springboot自动装配bean，声明队列和交换机

@Configuration
public class FanoutConfig 
    /**
     * 声明交换机
     * @return Fanout类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange()
        return new FanoutExchange("henrik.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);
    

publisher服务端

在publisher服务的SpringAmqpTest类中添加测试方法

@Test
public void testFanoutExchange() 
    // 队列名称
    String exchangeName = "henrik.fanout";
    // 消息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);

consumer服务端

在consumer服务的SpringRabbitListener中添加两个方法，作为消费者

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) 
    System.out.println("消费者1接收到Fanout消息：【" + msg + "】");


@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) 
    System.out.println("消费者2接收到Fanout消息：【" + msg + "】");

4.Routing 路由模式

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）
• 消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey
• Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的Routing key 与消息的 Routing key 完全一致，才会接收到消息
  图解：
• P：生产者，向 Exchange 发送消息，发送消息时，会指定一个routing key
• X：Exchange（交换机），接收生产者的消息，然后把消息递交给与 routing key 完全匹配的队列
• C1：消费者，其所在队列指定了需要 routing key 为 error 的消息
• C2：消费者，其所在队列指定了需要 routing key 为 info、error、warning 的消息

consumer服务端

基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。
在consumer的SpringRabbitListener中添加两个消费者，同时基于注解来声明队列和交换机

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


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

publisher服务端

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

5.Topics 通配符模式

• Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型Exchange 可以让队列在绑定 Routing key 的时候使用通配符
• Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
• 通配符规则：# 匹配一个或多个词，* 匹配不多不少恰好1个词，例如：item.# 能够匹配 item.insert.abc
  或者 item.insert，item.* 只能匹配 item.insert

• Queue1：绑定的是china.# ，因此凡是以 china.开头的routing key 都会被匹配到。包括china.news和china.weather
• Queue2：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配。包括china.news和japan.news

publisher服务端

/**
     * topicExchange
     */
@Test
public void testSendTopicExchange() 
    // 交换机名称
    String exchangeName = "henrik.topic";
    // 消息
    String message = "喜报！孙悟空大战哥斯拉，胜!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);

consumer服务端

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue1"),
    exchange = @Exchange(name = "henrik.topic", type = ExchangeTypes.TOPIC),
    key = "china.#"
))
public void listenTopicQueue1(String msg)
    System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");


@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue2"),
    exchange = @Exchange(name = "henrik.topic", type = ExchangeTypes.TOPIC),
    key = "#.news"
))
public void listenTopicQueue2(String msg)
    System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");

六.消息转换器

Spring会把发送的消息序列化为字节发送给MQ，接收消息的时候，会把字节反序列化为Java对象。只不过，默认情况下Spring采用的序列化方式是JDK序列化。JDK序列化存在下列问题：数据体积过大，有安全漏洞，可读性差。
所以JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。
①引入依赖

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

②配置消息转换器，在启动类中添加一个bean即可

@Bean
public MessageConverter jsonMessageConverter()
    return new Jackson2JsonMessageConverter();