文章目录

一、AMQP
二、Spring AMQP
- 2.1 介绍
- 2.2 SpringAMQP发送消息
- 2.3 SpringAMQP接收消息
- 2.4 WorkQueue模型
- - 2.4.1 概念
  - 2.4.2 示例
- 2.5 发布订阅模型
- - 2.5.1 介绍
  - 2.5.2 Fanout Exchange
  - 2.5.3 Direct Exchange
  - 2.5.4 Topic Exchange
- 2.6 消息转换器
- - 2.6.1 介绍
  - 2.6.2 切换消息转换器

一、AMQP

AMQP（高级消息队列协议）是一个标准的消息传递协议，用于在应用程序之间进行消息传递，特别是在分布式系统中。
在这里插入图片描述

AMQP提供了一个中间件消息传递机制，使不同的应用程序能够可靠地、安全地和高效地进行通信。它允许应用程序通过交换消息来进行通信，而不必直接进行网络通信。这种机制使得应用程序之间的通信更加灵活，因为它们可以独立地进行通信，而不必考虑其他应用程序的状态和可用性。

AMQP支持消息传递的各种场景，包括点对点通信、发布/订阅模式和请求/响应模式等。它具有诸如持久化消息、事务、消息确认和优先级等高级特性，使得它非常适合处理复杂的分布式应用程序。

二、Spring AMQP

2.1 介绍

Spring AMQP是基于Spring Framework的AMQP（高级消息队列协议）客户端库，用于在Java应用程序中使用AMQP进行消息传递。
在这里插入图片描述

Spring AMQP提供了一个高级的抽象层，使得开发人员可以很方便地使用AMQP进行消息传递，而不必直接处理AMQP的复杂性。它支持各种AMQP实现，例如RabbitMQ和Apache ActiveMQ等。

Spring AMQP提供了许多有用的特性，例如：

1.声明式配置：使用注释和Java配置声明式地配置消息交换和队列。

2.简化的消息发布和订阅：使用Spring AMQP，开发人员可以很方便地将消息发送到队列或从队列中接收消息。

3.异步处理：Spring AMQP提供了异步消息处理机制，可以轻松地处理大量消息。

4.异常处理：Spring AMQP提供了对异常处理的支持，以便在发生错误时进行适当的处理。

2.2 SpringAMQP发送消息

引入依赖：

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

在application.yml文件配置RabbitMQ的连接信息：

spring:
  rabbitmq:
  	#主机名
    host: 192.168.XX.XX
    #端口
    port: 5672
    #用户名
    username: caterpillar
    #密码
    password: 123456
    #虚拟主机
    virtual-host: /

调用RabbitTemplate的convertAndSend方法：

//自动注入RabbitTemplate
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void SendMessage(){
    //队列名称
    String queueName = "simple.queue";
    //所发送的消息
    String message = "hello,World!";
    //调用convertAndSend方法发送信息
    rabbitTemplate.convertAndSend(queueName,message);
}

2.3 SpringAMQP接收消息

在application.yml文件配置RabbitMQ的连接信息：

spring:
  rabbitmq:
  	#主机名
    host: 192.168.XX.XX
    #端口
    port: 5672
    #用户名
    username: caterpillar
    #密码
    password: 123456
    #虚拟主机
    virtual-host: /

在消费者的服务中新建一个类接收消息并处理

//声明为Bean
@Component
public class SpringRabbitListener {
    //指定消息队列的名称
    @RabbitListener(queues = "simple.queue")
    public void listenQueue(String msg){
        //处理指定消息队列所收到的信息
        System.out.println("收到的消息：" + msg);
    }
}

2.4 WorkQueue模型

2.4.1 概念

Work Queue（工作队列）模型是一种经典的消息队列模型，也被称为任务队列模型。它用于将耗时的任务分配给多个工作进程以便并行执行。
在这里插入图片描述

在Work Queue模型中，一个生产者将消息发送到一个队列中，多个消费者从队列中接收消息并处理它们。一个消息只能被一个消费者处理，即消息的消费是互斥的。

Work Queue模型的主要特点包括：

生产者将消息发送到队列中，而不是发送到特定的消费者。
消费者从队列中获取消息，并将其处理。
消息的处理是互斥的，即一个消息只能被一个消费者处理。
消费者可以平均分配任务，以便并行处理。

2.4.2 示例

生产者测试代码

@Test
public void testSendMessage2WorkQueue() throws InterruptedException {
    String queueName = "simple.queue";
    String message = "message -- ";
    //发送50次消息
    for (int i = 1; i <= 50; i++) {
        rabbitTemplate.convertAndSend(queueName, message + i);
        //每次循环延时20毫秒
        Thread.sleep(20);
    }
}

消费者代码

@Component
public class SpringRabbitListener {
	//消息队列1
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消息队列1收到的消息：" + msg + " -- " + LocalTime.now());
        Thread.sleep(20);
    }
	//消息队列2
    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        //使用err是为了输出时颜色不同便于区分
        System.err.println("消息队列2收到的消息：" + msg + " -- " + LocalTime.now());
        Thread.sleep(200);
    }
}

消费者配置添加

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1 #每次只预取一条消息，使消费能力的队列更强的做更多的接收

2.5 发布订阅模型

2.5.1 介绍

发布-订阅模型（Publish-Subscribe）是一种消息队列模型，主要用于将消息同时发给多个消费者。在发布-订阅模型中，消息的生产者将消息发送到一个交换机中，交换机会将消息同时发给多个与之绑定的队列，多个消费者可以分别从这些队列中获取消息并进行处理。
在这里插入图片描述

与点对点模型（P2P）不同，发布-订阅模型中的消息生产者并不需要知道消息的接收者，而是只需要将消息发送到交换机中即可。交换机会将消息广播给与之绑定的所有队列，多个消费者可以从不同的队列中获取并处理消息。消息的处理是并行的，多个消费者可以同时处理不同的消息，提高了系统的处理效率和吞吐量。

常见的exchange类型包括：

Fanout：广播
Direct：路由
Topic：话题

2.5.2 Fanout Exchange

Fanout Exchange是一种简单的Exchange类型，它会将所有收到的消息广播给与之绑定的所有队列，所有的队列都会收到相同的消息。在Fanout Exchange中，不需要指定Routing Key，消息会直接广播给所有的队列。Fanout Exchange通常用于需要将消息广播给多个消费者的场景，例如在线聊天室、实时数据分析等。
在这里插入图片描述

示例：

生产者测试代码

@Test
public void testSendFanoutExchange(){
    //交换机名称
    String exchangName = "caterpillar.fanout";
    //消息
    String message = "hello";
    //发送消息
    rabbitTemplate.convertAndSend(exchangName,"",message);
}

添加一个类作为消费者的配置类

//声明为一个配置
@Configuration
public class FanoutConfig {
    //声明交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("caterpillar.fanout");
    }
    //声明队列1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }
    //将队列1绑定到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }
    //声明队列2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }
    //将队列2绑定到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

消费者代码

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg){
        System.out.println("fanout1收到的消息：" + msg);
    }
    @RabbitListener(queues = "fanout.queue2")
    public void listenfanoutQueue(String msg){
        System.out.println("fanout2收到的消息：" + msg);
    }
}

2.5.3 Direct Exchange

Direct Exchange是一种常见的Exchange类型，它会将消息发送给与之匹配的队列。在Direct Exchange中，生产者将消息发送到指定的交换机中，并指定一个Routing Key，Exchange会将消息发送到与之绑定的Routing Key相同的队列中。Direct Exchange通常用于一些点对点的场景，例如订单处理、日志记录等。
在这里插入图片描述

示例：

生产者测试代码

@Test
public void testSendDirectExchange(){
    //交换机名称
    String exchangName = "caterpillar.direct";
    //消息
    String message = "hello,blue";
    //发送消息，只会发送给相同key的消费者，第二个参数是Routing Key
    rabbitTemplate.convertAndSend(exchangName,"red",message);
}

消费者代码

@RabbitListener(bindings = @QueueBinding(
	//声明队列
	value = @Queue(name = "direct.queue1"),
	//声明交换机
	exchange = @Exchange(name = "caterpillar.direct",type = ExchangeTypes.DIRECT),
	//声明绑定key
	key = {"red","blue"}
))
public void listenDirectQueue1(String msg){
	System.out.println("direct1收到的消息：" + msg);
}
@RabbitListener(bindings = @QueueBinding(
	//声明队列
	value = @Queue(name = "direct.queue2"),
	//声明交换机
	exchange = @Exchange(name = "caterpillar.direct",type = ExchangeTypes.DIRECT),
	//声明绑定key
	key = {"red","yellow"}
))
public void listenDirectQueue2(String msg){
	System.out.println("direct2收到的消息：" + msg);
}

在此示例中两个消费者都会收到消息，如果生产者指定的key为blue则只有direct1收到消息，如果生产者指定的key为yellow则只有direct2收到消息。

2.5.4 Topic Exchange

Topic Exchange是一种强大的Exchange类型，它会将消息发送到与之匹配的队列中。在Topic Exchange中，生产者可以指定一个带有通配符的Routing Key（*），Exchange会将消息发送到所有与之匹配的队列中。Topic Exchange支持通配符的Routing Key匹配，可以根据消息的主题进行路由，非常适合一些消息订阅和过滤的场景，例如新闻订阅、事件通知等。
在这里插入图片描述

Topic Exchange支持两种通配符：* 和 #。

*：表示匹配一个单词，可以用于匹配某个特定单词，例如"*.apple"可以匹配"green.apple"、“red.apple"等，但不能匹配"green.big.apple”。
#：表示匹配零个或多个单词，可以用于匹配某个前缀或者所有单词，例如"fruit.#“可以匹配"fruit.apple”、“fruit.orange”、"fruit.apple.red"等。

示例：

生产者测试代码

@Test
public void testSendTopicExchange(){
    //交换机名称
    String exchangName = "caterpillar.topic";
    //消息
    String message = "毛毛虫被网易招聘了！！！";
    //发送消息
    rabbitTemplate.convertAndSend(exchangName,"china.news",message);
}

消费者代码

@RabbitListener(bindings = @QueueBinding(
        //声明队列
        value = @Queue(name = "topic.queue1"),
        //声明交换机
        exchange = @Exchange(name = "caterpillar.topic",type = ExchangeTypes.TOPIC),
        //声明绑定key
        key = "china.#"
))
public void listenTopicQueue1(String msg){
    System.out.println("topic1收到的消息：" + msg);
}
@RabbitListener(bindings = @QueueBinding(
        //声明队列
        value = @Queue(name = "topic.queue2"),
        //声明交换机
        exchange = @Exchange(name = "caterpillar.topic",type = ExchangeTypes.TOPIC),
        //声明绑定key
        key = "#.news"
))
public void listenTopicQueue2(String msg){
    System.out.println("topic2收到的消息：" + msg);
}

在此示例中两个消费者都会收到消息，如果生产者指定的key为china.weather则只有topic1收到消息，如果生产者指定的key为Canada则只有topic2收到消息。

2.6 消息转换器

2.6.1 介绍

Spring AMQP提供了消息转换器（MessageConverter）的支持，用于将消息对象转换为字节数组或者将字节数组转换为消息对象，使得消息生产者和消费者之间可以方便地传输POJO对象。

Spring AMQP提供了以下几种消息转换器：

SimpleMessageConverter：默认的消息转换器，可以将Java对象转换为字节数组并发送到队列中，也可以从队列中接收字节数组并将其转换为Java对象。支持文本消息和字节消息的转换。
Jackson2JsonMessageConverter：基于Jackson库的消息转换器，可以将Java对象转换为JSON字符串并发送到队列中，也可以从队列中接收JSON字符串并将其转换为Java对象。支持文本消息和字节消息的转换。
Jackson2XmlMessageConverter：基于Jackson库的消息转换器，可以将Java对象转换为XML字符串并发送到队列中，也可以从队列中接收XML字符串并将其转换为Java对象。支持文本消息和字节消息的转换。
MarshallingMessageConverter：使用Spring的Marshaller和Unmarshaller进行消息转换的消息转换器。支持XML和Java Serialization格式。
ByteArrayMessageConverter：将Java对象转换为字节数组并发送到队列中，也可以从队列中接收字节数组并将其转换为Java对象。只支持字节消息的转换。

2.6.2 切换消息转换器

以切换为Jackson2JsonMessageConverter为例：

在主类pom文件引入依赖

<!--数据绑定模块-->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

在生产者和消费者的配置类中声明消息转换器的Bean

//消息转换器
@Bean
public MessageConverter messageConverter(){
    //Jackson消息转换工具
    return new Jackson2JsonMessageConverter();
}

消费者监听队列消息时接收的消息要和生产者中的类型一致

@RabbitListener(/*queues = 队列名称*/)
public void listenObjectQueue(/*消息类型 消息实例*/){
    //消息操作
}

一篇文章快速入门Spring AMQP

文章目录

一、AMQP

二、Spring AMQP

2.1 介绍

2.2 SpringAMQP发送消息

2.3 SpringAMQP接收消息

2.4 WorkQueue模型

2.4.1 概念

2.4.2 示例

2.5 发布订阅模型

2.5.1 介绍

2.5.2 Fanout Exchange

2.5.3 Direct Exchange

2.5.4 Topic Exchange

2.6 消息转换器

2.6.1 介绍

2.6.2 切换消息转换器

相关文章

重写equlas时为什么一定要重写hashcode方法？

webpack基本认知，它是什么，做什么的

DNS域名解析服务

ITSS服务经理、服务工程师线上开班在即

Qlik Sense 集合表达式详解

电子招标采购系统源码:营造全面规范安全的电子招投标环境，促进招投标市场健康可持续发展

CompletableFuture的基本使用和原理

如何将matlab的m文件转换成python文件

HttpWebRequest类

Spring MVC 接收 json 和返回 json （14）

2.微服务项目实战---环境搭建，实现电商中商品、订单、用户

【观察】构建“零信任”架构，筑起制造业安全“护城河”

使用全球融合CDN的10大优势

32道子网划分练习题详细解析含答案

企业云成本优化：减少企业云支出的终极指南

【Linux网络服务】DNS域名解析服务服务

[ARM+Linux] 基于wiringPi库的串口通信

JavaSE-part1

服务器初始化

数据结构与算法——深度寻路算法