MQ之Spring AMQP学习

2024-03-20 04:20
文章标签 java 学习 spring amqp mq

本文主要是介绍MQ之Spring AMQP学习，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Spring AMQP学习

Spring AMQP

AMQP是Advanced Message Queuing Protocol的缩写。AMQP是用于在应用程序之间传递消息的开放标准，该协议与语言和平台无关，更符合微服务中独立性的要求。
Spring AMQP是基于AMQP协议定义的一套API规范，提供了模版来发送和接收消息，包含两部分，其中spring-amqp是基础抽象，spring-rabbit是底层的默认实现。

SpringAMQP提供了三个功能：

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

Spring AMQP使用

首先需要引入依赖

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

RabbitMQ配置：

spring:rabbitmq:port: 5672  # mq server 端口号，注意：部署15672virtual-host: /test  #虚拟主机host: 192.168.200.215  # mq server ipusername: testuser1   # mq管理后台用户名password: testuser1  # mq管理后台密码

简单队列

work queue（工作队列）

work queue简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。
当消息处理比较耗时的时候，可能生产消息的速度会远远大于消费的速度。如果一直这样，消息就会堆积越来越多，无法及时处理。
此时就可以使用work模型，多个消费者共同处理消息，这样消费速度就能大大提高了。

work queu默认是采用预取模式去消费消息的，也就是如果有多个消费者，则会先预分配给每个消费一个消息，消费完以后，再给每个消费者继续分配。

消息生产代码如下：

    @RequestMapping("/work-queue")public String testWorkQueue() {// 消息String message = "hello, spring amqp!";for (int i = 0; i < 50; i++) {// 发送消息rabbitTemplate.convertAndSend(SIMPLE_QUEUE_NAME, message + i);}log.info("消息发送成功: {}", message);return "success";}

消费者代码如下：

@Component
@Slf4j
public class SimpleQueueListener {@RabbitListener(queues = "simple.queue")public void listenerWorkQueue1(String message) {log.info("消费者1接收到的消息: {}", message);Thread.sleep(5);}@RabbitListener(queues = "simple.queue")public void listenerWorkQueue2(String message) {log.info("消费者2接收到的消息: {}", message);Thread.sleep(200);}}

结果如下图所示，可以看到两个消费者消费消息是均匀分配的。
在这里插入图片描述

如果想实现“能者多劳”，消费消息快的消费者多消费，可以使用下面的配置

  rabbitmq:host: 192.168.200.215username: testuser1password: testuser1port: 5672  #这里的端口是5762，不是15762virtual-host: /testlistener:simple:prefetch: 1 #每次只能得到一条消息，处理完ACK之后，才能获取下一个消息

加入配置以后，可以看到消费者2消费的消息比消费者1多。
在这里插入图片描述

发布订阅

发布订阅的模型如下图所示：
在这里插入图片描述

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

  • Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
  • Exchange：交换机，图中的X。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：
    • Fanout：广播，将消息交给所有绑定到交换机的队列
    • Direct：定向，把消息交给符合指定routing key 的队列
    • Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
  • Consumer：消费者，与以前一样，订阅队列，没有变化
  • Queue：消息队列也与以前一样，接收消息、缓存消息
Fanout模式示例

声明交换机，绑定队列和交换机。这个代码需要和消息发送者在同一个服务中。

@Configuration
public class FanoutConfiguration {/*** 声明交换机**/@Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange("demo.fanout");}@Beanpublic Queue fanoutQueue1() {return new Queue("fanout.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1(@Autowired Queue fanoutQueue1, @Autowired FanoutExchange fanoutExchange) {return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2个队列*/@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}}

发送消息。注意：convertAndSend方法中指定了交换机的名称

    @GetMapping("/fanout")public String testFanoutQueue() {// 消息String message = "hello, fanout queue!";//交换机名称String exchange  = "demo.fanout";rabbitTemplate.convertAndSend(exchange, "", message);log.info("消息发送成功: {}", message);return "success";}

消费者接收消息，接收消息和前面简单队列是一样的。

@Component
@Slf4j
public class SimpleQueueListener {@RabbitListener(queues = "fanout.queue1")public void fanoutQueue1(String message) throws InterruptedException {log.info("消费者1接收到的消息: {}", message);}@RabbitListener(queues = "fanout.queue2")public void fanoutQueue2(String message) throws InterruptedException {log.info("消费者2接收到的消息: {}", message);}}

结果如下图所示：
在这里插入图片描述

Topic模式

在fanout模式下，一条消息会被所有订阅的队列消费，但是有时候，我们想让消息被不同的队列消费，这时候就需要用到direct类型的Exchange。
在这里插入图片描述

消息监听，这里使用的注解的方式，使用@Queue指定队列，@Exchange指定交换机，key指定路由的key，数组类型，可以指定多个

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"),exchange = @Exchange(name = "direct", type = ExchangeTypes.DIRECT),key = {"blue"}))public void listenDirectQueue1(String message) {log.info("消费者1接收到的消息: {}", message);}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"),exchange = @Exchange(name = "direct", type = ExchangeTypes.DIRECT), key = {"yellow"}))public void listenDirectQueue2(String message) {log.info("消费者2接收到的消息: {}", message);}

发送消息，发送消息时需要指定路由key，这样消息才能被指定的队列收到

        // 消息String message = "hello, direct queue!";//交换机名称String exchange  = "direct";rabbitTemplate.convertAndSend(exchange, "yellow", message);
Topic模式

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

接收消息监听

    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "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 = "topic", type = ExchangeTypes.TOPIC),key = "#.news"))public void listenTopicQueue2(String msg){System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");}

发送消息，需要匹配通配符，这样才能收到消息

        // 消息String message = "hello, topic queue!";//交换机名称String exchange  = "topic";rabbitTemplate.convertAndSend(exchange, "chiana.news", message);

结果如下：

在这里插入图片描述

参考

  • Spring AMQP 官方文档

这篇关于MQ之Spring AMQP学习的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/828264。

相关文章

51单片机学习记录———定时器

51单片机学习记录———定时器

文章目录 前言一、定时器介绍二、STC89C52定时器资源三、定时器框图四、定时器模式五、定时器相关寄存器六、定时器练习 前言 一个学习嵌入式的小白~ 有问题评论区或私信指出~ 提示：以下是本篇文章正文内容，下面案例可供参考 一、定时器介绍 定时器介绍：51单片机的定时器属于单片机的内部资源，其电路的连接和运转均在单片机内部完成。 定时器作用： 1.用于计数系统，可
阅读更多...
问题：第一次世界大战的起止时间是 #其他#学习方法#微信

问题：第一次世界大战的起止时间是 #其他#学习方法#微信

问题：第一次世界大战的起止时间是 A.1913 ~1918 年 B.1913 ~1918 年 C.1914 ~1918 年 D.1914 ~1919 年 参考答案如图所示
阅读更多...
Java五子棋之坐标校正

Java五子棋之坐标校正

上篇针对了Java项目中的解构思维，在这篇内容中我们不妨从整体项目中拆解拿出一个非常重要的五子棋逻辑实现：坐标校正，我们如何使漫无目的鼠标点击变得有序化和可控化呢？ 目录 一、从鼠标监听到获取坐标 1.MouseListener和MouseAdapter 2.mousePressed方法 二、坐标校正的具体实现方法 1.关于fillOval方法 2.坐标获取 3.坐标转换 4.坐
阅读更多...
Spring Cloud：构建分布式系统的利器

Spring Cloud：构建分布式系统的利器

引言 在当今的云计算和微服务架构时代，构建高效、可靠的分布式系统成为软件开发的重要任务。Spring Cloud 提供了一套完整的解决方案，帮助开发者快速构建分布式系统中的一些常见模式（例如配置管理、服务发现、断路器等）。本文将探讨 Spring Cloud 的定义、核心组件、应用场景以及未来的发展趋势。 什么是 Spring Cloud Spring Cloud 是一个基于 Spring
阅读更多...
[word] word设置上标快捷键 #学习方法#其他#媒体

[word] word设置上标快捷键 #学习方法#其他#媒体

word设置上标快捷键 办公中，少不了使用word，这个是大家必备的软件，今天给大家分享word设置上标快捷键，希望在办公中能帮到您！ 1、添加上标 在录入一些公式，或者是化学产品时，需要添加上标内容，按下快捷键Ctrl+shift++就能将需要的内容设置为上标符号。 word设置上标快捷键的方法就是以上内容了，需要的小伙伴都可以试一试呢！
阅读更多...
AssetBundle学习笔记

AssetBundle学习笔记

AssetBundle是unity自定义的资源格式，通过调用引擎的资源打包接口对资源进行打包成.assetbundle格式的资源包。本文介绍了AssetBundle的生成，使用，加载，卸载以及Unity资源更新的一个基本步骤。 目录 1.定义： 2.AssetBundle的生成： 1）设置AssetBundle包的属性——通过编辑器界面 补充：分组策略 2）调用引擎接口API
阅读更多...
Javascript高级程序设计(第四版)--学习记录之变量、内存

Javascript高级程序设计(第四版)--学习记录之变量、内存

原始值与引用值 原始值：简单的数据即基础数据类型，按值访问。 引用值：由多个值构成的对象即复杂数据类型，按引用访问。 动态属性 对于引用值而言，可以随时添加、修改和删除其属性和方法。 let person = new Object();person.name = 'Jason';person.age = 42;console.log(person.name,person.age);//'J
阅读更多...
java8的新特性之一（Java Lambda表达式）

java8的新特性之一（Java Lambda表达式）

1：Java8的新特性 Lambda 表达式： 允许以更简洁的方式表示匿名函数（或称为闭包）。可以将Lambda表达式作为参数传递给方法或赋值给函数式接口类型的变量。 Stream API： 提供了一种处理集合数据的流式处理方式，支持函数式编程风格。 允许以声明性方式处理数据集合（如List、Set等）。提供了一系列操作，如map、filter、reduce等，以支持复杂的查询和转
阅读更多...
大学湖北中医药大学法医学试题及答案，分享几个实用搜题和学习工具 #微信#学习方法#职场发展

大学湖北中医药大学法医学试题及答案，分享几个实用搜题和学习工具 #微信#学习方法#职场发展

今天分享拥有拍照搜题、文字搜题、语音搜题、多重搜题等搜题模式，可以快速查找问题解析，加深对题目答案的理解。 1.快练题 这是一个网站 找题的网站海量题库,在线搜题,快速刷题~为您提供百万优质题库,直接搜索题库名称,支持多种刷题模式:顺序练习、语音听题、本地搜题、顺序阅读、模拟考试、组卷考试、赶快下载吧! 2.彩虹搜题 这是个老公众号了 支持手写输入，截图搜题，详细步骤，解题必备
阅读更多...
Java面试八股之怎么通过Java程序判断JVM是32位还是64位

Java面试八股之怎么通过Java程序判断JVM是32位还是64位

怎么通过Java程序判断JVM是32位还是64位 可以通过Java程序内部检查系统属性来判断当前运行的JVM是32位还是64位。以下是一个简单的方法： public class JvmBitCheck {public static void main(String[] args) {String arch = System.getProperty("os.arch");String dataM
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2