微服务中间件--MQ

news2024/12/26 3:09:52

MQ

  • MQ
    • a.安装RabbitMQ
    • b.消息模型
    • c.SpringAMQP发送和接收
    • d.WorkQueue模型
    • e.发布订阅模型
      • 1) FanoutExchange
      • 2) DirectExchange
      • 3) TopicExchange
    • f.消息转换器

MQ

同步调用的问题

微服务间基于Feign的调用就属于同步方式，存在一些问题。

  • 耦合度高：每次加入新的需求，都要修改原来的代码
  • 性能下降：调用者需要等待服务提供者响应，如果调用链过长则响应时间等于每次调用的时间之和。
  • 资源浪费：调用链中的每个服务在等待响应过程中，不能释放请求占用的资源，高并发场景下会极度浪费系统资源
  • 级联失败：如果服务提供者出现问题，所有调用方都会跟着出问题，如同多米诺骨牌一样，迅速导致整个微服务群故障

异步调用方案

异步调用常见实现就是事件驱动模式

  • 耦合度低
  • 吞吐量提升
  • 故障隔离
  • 流量削峰

异步通信的缺点：

  • 依赖于Broker的可靠性、安全性、吞吐能力
  • 架构复杂了，业务没有明显的流程线，不好追踪管理

什么是MQ

MQ （MessageQueue），中文是消息队列，字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。

在这里插入图片描述

a.安装RabbitMQ

1.在线拉取

docker pull rabbitmq:3-management

2.执行下面的命令来运行MQ容器：

docker run \
 -e RABBITMQ_DEFAULT_USER=root \
 -e RABBITMQ_DEFAULT_PASS=root \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management

15672 控制台端口号

5672 消息通信端口

在这里插入图片描述

RabbitMQ中的几个概念：

  • channel：操作MQ的工具
  • exchange：路由消息到队列中
  • queue：缓存消息
  • virtual host：虚拟主机，是对queue、exchange等资源的逻辑分组

b.消息模型

MQ的官方文档中给出了5个MQ的Demo示例，对应了几种不同的用法：

  • 基本消息队列（BasicQueue）
  • 工作消息队列（WorkQueue）
  • 发布订阅（Publish、Subscribe），又根据交换机类型不同分为三种：
    • Fanout Exchange：广播
    • Direct Exchange：路由
    • Topic Exchange：主题

官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色：

  • publisher：消息发布者，将消息发送到队列
  • queuequeue：消息队列，负责接受并缓存消息
  • consumer：订阅队列，处理队列中的消息

基本消息队列的消息发送流程：

1.建立connection

2.创建channel

3.利用channel声明队列

4.利用channel向队列发送消息

基本消息队列的消息接收流程：

1.建立connection

2.创建channel

3.利用channel声明队列 (避免队列不存在，所以双方都声明)

4.定义consumer的消费行为handleDelivery()

5.利用channel将消费者与队列绑定

c.SpringAMQP发送和接收

案例：利用SpringAMQP实现HelloWorld中的基础消息队列功能

流程如下：

1.在父工程中引入spring-amqp的依赖

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

2.在publisher中编写测试方法，向simple.queue发送消息

  • 在publisher服务中编写application.yml，添加mq连接信息：
spring:
  rabbitmq:
    host: 192.168.58.128 # rabbitMq的ip地址
    port: 5672 # 端口
    username: root
    password: root
    virtual-host: /
  • 在publisher服务中新建一个测试类，编写测试方法：
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSendMessage2SimpleQueue(){
        String queueName = "simple.queue";
        String message = "Hello, spring amqp!";
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

3.在consumer中编写消费逻辑，监听simple.queue

  • 在consumer服务中编写application.yml，添加mq连接信息：
spring:
  rabbitmq:
    host: 192.168.58.128
    port: 5672
    username: root
    password: root
    virtual-host: /
  • 在consumer服务中新建一个类，编写消费逻辑 (添加监听注解**@RabbitListener**)：
@Component
public class SpringRabbitListener {
    @RabbitListener(queues = {"simple.queue"})
    public void listenSimpleQueue(String msg){
        System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
    }
}

d.WorkQueue模型

Work queue，工作队列，可以提高消息处理速度，避免队列消息堆积

在这里插入图片描述

案例：模拟WorkQueue，实现一个队列绑定多个消费者

1.在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue

@Test
public void testSendMessage2WorkQueue() throws Exception {
    String queueName = "simple.queue";
    String message = "Hello, message__";

    for (int i = 1; i <= 50; i++) {
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}

2.在consumer服务中定义两个消息监听者，都监听simple.queue队列

3.消费者1每秒处理50条消息，消费者2每秒处理10条消息

@Component
public class SpringRabbitListener {

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

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

消费预取限制

修改application.yml文件，设置preFetch这个值，可以控制预取消息的上限：

spring:
  rabbitmq:
    host: 192.168.58.128
    port: 5672
    username: root
    password: root
    virtual-host: /
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

可以提高整个消息队列的速度，避免消息堆积

e.发布订阅模型

发布（ Publish ）、订阅（ Subscribe ）

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。

常见exchange类型包括：

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

在这里插入图片描述

注意：exchange负责消息路由，而不是存储，路由失败则消息丢失

1) FanoutExchange

Fanout Exchange 会将接收到的消息路由给每一个跟其绑定的queue

在这里插入图片描述

案例：利用SpringAMQP演示FanoutExchange的使用

1.在consumer服务中，利用代码声明队列、交换机，并将两者绑定

在consumer服务常见一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding

@Configuration
public class FanoutConfig {
    // abcd.fanout
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("abcd.fanout");
    }

    // fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

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

    // fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

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

2.在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2

@Component
public class SpringRabbitListener {
    @RabbitListener(queues = {"fanout.queue1"})
    public void listenFanoutQueue1(String msg){
        System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】");
    }

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

3.在publisher中编写测试方法，向abcd.fanout发送消息

@Test
public void testSendFanoutExchange() {
   // 交换机名称
   String exchangeName = "abcd.fanout";

   //消息
   String message = "Hello, every one!";

   // 发送消息: 参数依次为：交换机名称，RoutingKey，消息
   rabbitTemplate.convertAndSend(exchangeName, "", message);
}

2) DirectExchange

Direct Exchange 会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes）。

  • 每一个Queue都与Exchange设置一个BindingKey
  • 发布者发送消息时，指定消息的RoutingKey
  • Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

在这里插入图片描述

案例：利用SpringAMQP演示DirectExchange的使用

1.利用@RabbitListener声明Exchange、Queue、RoutingKey

2.在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2

@Component
public class SpringRabbitListener {

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

3.在publisher中编写测试方法，向abcd. direct发送消息

@Test
public void testSendDirectExchange(){
    // 交换机名称
    String exchangeName = "abcd.direct";

    // 消息
    String message = "Hello, blue!";

    // 发送消息，参数依次为：交换机名称，RoutingKey，消息
    rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

3) TopicExchange

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以. 分割。Queue与Exchange指定BindingKey时可以使用通配符：

  • #：代指0个或多个单词
  • *：代指一个单词

在这里插入图片描述

案例：利用SpringAMQP演示TopicExchange的使用

1.利用@RabbitListener声明Exchange、Queue、RoutingKey

2.在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2

@Component
public class SpringRabbitListener {

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

3.在publisher中编写测试方法，向abcd. topic发送消息

@Test
public void testSendTopicExchange(){
    // 交换机名称
    String exchangeName = "abcd.topic";

    // 消息
    String message = "Hello, new!";

    // 发送消息，参数依次为：交换机名称，RoutingKey，消息
    rabbitTemplate.convertAndSend(exchangeName, "hello.news", message);
}

f.消息转换器

在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。

案列：测试发送Object类型消息

1.在consumer中利用@Bean声明一个队列：

@Configuration
public class FanoutConfig {
    @Bean
    public Queue objectQueue() {
        return new Queue("object.queue");
    }
}

2.在publisher中发送消息以测试：

@Test
public void testSendObjectQueue(){
    Map<String, Object> msg = new HashMap<>();
    msg.put("name", "张三");
    msg.put("age", 21);
    rabbitTemplate.convertAndSend("object.queue", msg);
}

在这里插入图片描述

消息转换器

Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。如果要修改只需要定义一个MessageConverter 类型的Bean即可。推荐用JSON方式序列化

1.在publisher服务引入依赖

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

2.在publisher服务声明MessageConverter：

@Bean
public MessageConverter messageConverter(){
    return new Jackson2JsonMessageConverter();
}

在这里插入图片描述

3.在consumer服务引入Jackson依赖：

  • 与步骤1一样

4.在consumer服务定义MessageConverter：

  • 与步骤2一样

5.定义一个消费者，监听object.queue队列并消费消息：

@RabbitListener(queues = "object.queue")
public void listenObjectQueue(Map<String, Object> msg){
    System.out.println("接收到object.queue的消息：" + msg);
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/919470.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【网络安全】防火墙知识点全面图解（二）

【网络安全】防火墙知识点全面图解（二）

本系列文章包含&#xff1a; 【网络安全】防火墙知识点全面图解&#xff08;一&#xff09;【网络安全】防火墙知识点全面图解&#xff08;二&#xff09; 防火墙知识点全面图解&#xff08;二&#xff09; 21、路由器的访问控制列表是什么样的&#xff1f;22、防火墙的安全策…
阅读更多...
java电子病历源码 电子病历编辑器源码 病历在线制作、管理和使用

java电子病历源码 电子病历编辑器源码 病历在线制作、管理和使用

电子病历在线制作、管理和使用的一体化电子病历解决方案&#xff0c;通过一体化的设计&#xff0c;提供对住院病人的电子病历书写、保存、修改、打印等功能。电子病历系统将临床医护需要的诊疗资料以符合临床思维的方法展示。建立以病人为中心&#xff0c;以临床诊疗信息为主线…
阅读更多...
mongoDB的CRUD

mongoDB的CRUD

阅读更多...
【WebSocket】前端使用WebSocket实时通信

【WebSocket】前端使用WebSocket实时通信

目录 前言什么是WebSocketWebSocket的工作原理WebSocket与HTTP的关系HTTP建立持久化连接WebSocket类封装 前言 最近写项目&#xff0c;需要实现消息通知和实时聊天的功能&#xff0c;就去了解了一些关于websocket的知识&#xff0c;总结如下。 什么是WebSocket WebSocket 是一…
阅读更多...
Prometheus 监控系统

Prometheus 监控系统

常用的监控系统有哪些&#xff1f; 老牌传统 Zabbix Nagios Cacti 新一代的 Prometheus 夜莺 Zabbix 和 Prometheus 的区别&#xff1f;如何选择&#xff1f;【重中之重】 Zabbix 更适用于传统业务架构的物理机、虚拟机环境的监控&#xff0c;对容器环境的支持较差&#xf…
阅读更多...
5.从头跑一个pipeline

5.从头跑一个pipeline

1.安装torch pip install torchvision torch PyTorch的torchvision.models模块中自带的很多预定义模型。torchvision 是PyTorch的一个官方库&#xff0c;专门用于处理计算机视觉任务。在这个库中&#xff0c;可以找到许多常用的卷积神经网络模型&#xff0c;包括ResNet、VGG、…
阅读更多...
【Eclipse】汉化简体中文教程（官方汉化包，IDE自带软件安装功能），图文详情

【Eclipse】汉化简体中文教程（官方汉化包，IDE自带软件安装功能），图文详情

目录 0.环境 1.步骤 1&#xff09;查看eclipse的版本 2&#xff09;在官网找语言包&#xff0c;并复制链接 3&#xff09;将链接复制到eclipse中 4&#xff09;汉化完成 0.环境 windows11&#xff0c;64位&#xff1b; eclipse 2021-6版本 1.步骤 思路&#xff1a;在官网找…
阅读更多...
【FAQ】云存储EasyCVR视频汇聚平台分发rtsp流时，出现“用户已过期”提示该如何解决？

【FAQ】云存储EasyCVR视频汇聚平台分发rtsp流时，出现“用户已过期”提示该如何解决？

视频云存储/安防监控EasyCVR视频汇聚平台基于云边端智能协同&#xff0c;支持海量视频的轻量化接入与汇聚、转码与处理、全网智能分发、视频集中存储等。音视频流媒体视频平台EasyCVR拓展性强&#xff0c;视频能力丰富&#xff0c;具体可实现视频监控直播、视频轮播、视频录像、…
阅读更多...
2023-8-23 合并集合

2023-8-23 合并集合

题目链接&#xff1a;合并集合 #include <iostream>using namespace std;const int N 100010;int n, m; int p[N];int find(int x) {if(p[x] ! x) p[x] find(p[x]);return p[x]; }int main() {cin >> n >> m;for(int i 1; i < n; i) p[i] i;while(m…
阅读更多...
webrtc的Sdp中的Plan-b和UnifiedPlan

webrtc的Sdp中的Plan-b和UnifiedPlan

在一些类似于视频会议场景下&#xff0c;媒体会话参与者需要接收或者发送多个流&#xff0c;例如一个源端&#xff0c;同时发送多个左右音轨的音频&#xff0c;或者多个摄像头的视频流&#xff1b;在2013年&#xff0c;提出了2个不同的SDP IETF草案Plan B和Unified Plan&#x…
阅读更多...
云服务器（Centos7系统）配置JAVA+mysql+tomcat 环境

云服务器（Centos7系统）配置JAVA+mysql+tomcat 环境

文章主要内容来源云服务器&#xff08;Centos7系统&#xff09;部署javaweb项目&#xff08;二&#xff09;配置JAVAmysqltomcat 环境_man_zuo的博客-CSDN博客 模仿途中遇到的问题 连接无效 有时连接无法下载&#xff0c;可能是过期了&#xff0c;将其更换为官网给的下载连接即…
阅读更多...
精准高效农业作业，植保无人机显身手

精准高效农业作业，植保无人机显身手

中国作为农业大国&#xff0c;拥有约18亿亩的农田&#xff0c;每年都需要进行种子喷洒和农药施用等农业作业&#xff0c;对于普通农户来说&#xff0c;这是一项耗时耗力的工程&#xff0c;同时&#xff0c;人工喷洒农药极易造成农药慢性中毒&#xff0c;对农民的身体健康产生极…
阅读更多...
Linux 虚拟机安装 hadoop

Linux 虚拟机安装 hadoop

目录 1 hadoop下载 2 解压hadoop 3 为 hadoop 文件夹改名 4 给 hadoop 文件夹赋权 5 修改环境变量 6 刷新环境变量 7 在hadoop313目录下创建文件夹data 8 检查文件 9 编辑 ./core-site.xml文件 10 编辑./hadoop-env.sh文件 11 编辑./hdfs-site.xml文件 12 编辑./mapr…
阅读更多...
python并发编程

python并发编程

一、程序提速的方法 二、python对并发编程的支持 多线程&#xff1a;threading&#xff0c;利用CPU和IO可以同时执行的原理&#xff0c;让CPU不会干巴巴等待IO完成&#xff1b;多进程&#xff1a;multiprocess&#xff0c;利用多核CPU的能力&#xff0c;真正的并行执行任务&am…
阅读更多...
大数据Flink（六十六）：Flink的重要概念和小结

大数据Flink（六十六）：Flink的重要概念和小结

文章目录 Flink的重要概念和小结 一、​​​​​​​​​​​​​​数据流图（Dataflow Graph）
阅读更多...
三维模拟推演电子沙盘虚拟数字沙盘开发教程第13课

三维模拟推演电子沙盘虚拟数字沙盘开发教程第13课

三维模拟推演电子沙盘虚拟数字沙盘开发教程第13课 该数据库中只提供 成都市火车南站附近的数据请注意&#xff0c;104.0648,30.61658 在SDK中为了方便三方数据的接入&#xff0c;引入了一个用户层接口。主要是完成三方数据的接入&#xff0c;含动态数据&#xff08;如GPS&…
阅读更多...
Shopee卖家开店后用知虾shopee抢占市场！

Shopee卖家开店后用知虾shopee抢占市场！

​ Shopee卖家在实际操作时需要根据自己的情况和行业特点进行具体的操作和决策&#xff0c;这里分享知虾shopee运营方法&#xff1a; 市场研究&#xff1a;了解知虾shopee平台上的市场趋势、受众需求和竞争情况。观察热门品类、畅销产品以及受欢迎的品牌&#xff0c;找到有潜力…
阅读更多...
如何使用CSS实现一个水平居中和垂直居中的布局？

如何使用CSS实现一个水平居中和垂直居中的布局？

聚沙成塔每天进步一点点 ⭐ 专栏简介⭐ 水平居中布局⭐ 垂直居中布局⭐ 写在最后 ⭐ 专栏简介 前端入门之旅&#xff1a;探索Web开发的奇妙世界 记得点击上方或者右侧链接订阅本专栏哦 几何带你启航前端之旅 欢迎来到前端入门之旅&#xff01;这个专栏是为那些对Web开发感兴趣…
阅读更多...
设计模式（3）抽象工厂模式

设计模式（3）抽象工厂模式

一、概述&#xff1a; 1、提供一个创建一系列相关或相互依赖对象的接口&#xff0c;而无须指定它们具体的类。 2、结构图&#xff1a; 3、举例代码&#xff1a; &#xff08;1&#xff09; 实体&#xff1a; public interface IUser {public void insert(User user);public…
阅读更多...
android studio gradle build running慢 卡住不动 失败 原因与解决方式

android studio gradle build running慢 卡住不动 失败 原因与解决方式

快速导航 分析原因解决办法 分析原因 主要原因是 gradle 构建时无法从网络获取需要的包或库。 解决办法 将国外库替换为阿里云镜像库。 例如 google 对应的库是 maven { url ‘https://maven.aliyun.com/repository/google’ }
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023