微服务中间件--MQ

这篇具有很好参考价值的文章主要介绍了微服务中间件--MQ。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

MQ

同步调用的问题

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

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

异步调用方案

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

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

异步通信的缺点:

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

什么是MQ

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

微服务中间件--MQ,微服务中间件,微服务,中间件,架构,java,spring cloud

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 消息通信端口

微服务中间件--MQ,微服务中间件,微服务,中间件,架构,java,spring cloud

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,工作队列,可以提高消息处理速度,避免队列消息堆积

微服务中间件--MQ,微服务中间件,微服务,中间件,架构,java,spring cloud

案例:模拟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:话题

微服务中间件--MQ,微服务中间件,微服务,中间件,架构,java,spring cloud

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

1) FanoutExchange

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

微服务中间件--MQ,微服务中间件,微服务,中间件,架构,java,spring cloud

案例:利用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一致的队列

微服务中间件--MQ,微服务中间件,微服务,中间件,架构,java,spring cloud

案例:利用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个或多个单词
  • *:代指一个单词

微服务中间件--MQ,微服务中间件,微服务,中间件,架构,java,spring cloud

案例:利用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);
}

微服务中间件--MQ,微服务中间件,微服务,中间件,架构,java,spring cloud

消息转换器

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();
}

微服务中间件--MQ,微服务中间件,微服务,中间件,架构,java,spring cloud

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

  • 与步骤1一样

4.在consumer服务定义MessageConverter:

  • 与步骤2一样

5.定义一个消费者,监听object.queue队列并消费消息:文章来源地址https://www.toymoban.com/news/detail-672603.html

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

到了这里,关于微服务中间件--MQ的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Spring Cloud 项目中实现推送消息到 RabbitMQ 消息中间件

    (注:安装在虚拟机则填虚拟机地址,否则则为本机地址) 用户名和密码都为guest 看到如下页面则为RabbitMQ安装登录成功。 三、依赖注入 导入依赖坐标 四、配置yaml文件 配置yaml配置文件 (注:host为地址,如果安装在虚拟机则为虚拟机地址,安装在本机则本机地址。port为端

    2024年04月13日
    浏览(57)
  • 实战:Spring Cloud Stream集成兼容多消息中间件kafka、rabbitmq

    前面的博文我们介绍并实战演示了Spring Cloud Stream整合rabbitmq,其中主要介绍了如何使用和配置完成消息中间件的集成。但是,在实际的生产环境中可能会用到多个消息中间件,又或者是由于业务改变需要更换消息中间件,在这些情况下我们的Spring Cloud Stream框架可以完全兼容多

    2024年02月08日
    浏览(54)
  • Spring Cloud Alibaba 最新版本(基于Spring Boot 3.1.0)整合完整使用及与各中间件集成

    目录 前言 源码地址 官方中文文档 使用版本 spring Spring Boot 3.1.0 中间件 使用到的组件与功能 环境安装 虚拟机 nexus nacos 集成过程 工程搭建 父工程搭建 子工程 服务集成 nacos集成 配置文件 服务注册与发现-discovery 服务注册 启动 服务发现 测试 配置管理-config 新增配置  测试

    2024年02月07日
    浏览(54)
  • 【消息中间件】详解mq消息积压

    作者简介 目录 1.产生原因 2.解决办法 2.1.事前处理机制 2.2.事中处理机制 2.3.事后处理机制 消息积压(Message Backlog)指的是在消息队列(MQ)系统中等待被处理的消息数量超过了正常的处理速度,导致消息在队列中积压堆积的情况。 消息积压的常见表现: 系统资源使用率上升

    2024年02月07日
    浏览(47)
  • spring cloud 搭建消息中间件 RabbitMQ 环境、Mac/Windows下载安装RabbitMQ、配置RabbitMQ环境变量

    spring boot、spring cloud工程:Mac/Windows下载安装Erlang、RabbitMQ,并配置环境变量。 这里学习如何安装 RabbitMQ,因为远程配置中心的动态更新需要结合 RabbitMQ 来使用。 这里给出自己下载和使用的百度网盘链接:Erlang 版本为25.3.2、RabbitMQ版本为3.12.1 : 链接:百度网盘链接 提取码:

    2024年02月15日
    浏览(70)
  • 快速掌握MQ消息中间件rabbitmq

    Survive by day and develop by night. talk for import biz , show your perfect code,full busy,skip hardness,make a better result,wait for change,challenge Survive. happy for hardess to solve denpendies. 需求: 1.video A https://www.bilibili.com/video/BV1cb4y1o7zz?p=12vd_source=533ee415c42b820b0f4105acb4932a02 参考资料 官方文档 开源社区 博客文

    2024年02月11日
    浏览(47)
  • java后端技术汇总 + 中间件 + 架构思想

    1. 华为OD机考题 + 答案 2023华为OD统一考试(A+B卷)题库清单-带答案(持续更新) 2023年华为OD真题机考题库大全-带答案(持续更新) 2. 面试题 一手真实java面试题:2023年各大公司java面试真题汇总--持续更新 3. 技术知识 java后端技术汇总 + 中间件 + 架构思想 类型 难度 Spring、

    2024年02月13日
    浏览(76)
  • 【中间件】MQ是什么?RabbitMQ又是什么?

    大家好,我是卷心菜。如果您看完这篇文章有所收获,可以三连支持博主哦~,嘻嘻。 🎁 作者简介: 在校大学生一枚,Java领域新星创作者,Java、Python正在学习中,期待和大家一起学习一起进步~ 💗 个人主页:我是一棵卷心菜的个人主页 🔶 本文专栏:RabbitMQ学习 📕 自我提

    2024年02月02日
    浏览(48)
  • MQ(消息中间件)概述及 RabbitMQ 的基本介绍

    消息队列中间件是分布式系统中重要的组件,主要解决 应用解耦,异步消息,流量削锋等 问题,实现高性能,高可用,可伸缩和最终一致性架构。流量削锋 : 削减峰值压力(秒杀,抢购) MQ(Message Queue,消息队列)是典型的生产者、消费者模型。生产者不断向消息队列中

    2024年02月12日
    浏览(47)
  • 【消息中间件】详解三大MQ:RabbitMQ、RocketMQ、Kafka

    作者简介 前言 博主之前写过一个完整的MQ系列,包含RabbitMQ、RocketMQ、Kafka,从安装使用到底层机制、原理。专栏地址: https://blog.csdn.net/joker_zjn/category_12142400.html?spm=1001.2014.3001.5482 本文是该系列的清单综述,会拉通来聊一下三大MQ的特点和各种适合的场景。 目录 1.概述 1.1.M

    2024年02月09日
    浏览(53)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包