服务器的异步通信——RabbitMQ

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

目录

一、同步通信 VS 异步通信

二、MQ——消息队列

RabbitMQ 

RabbitMQ安装 

RabbitMQ的整体架构

常见消息模型 

 基本消息队列(BasicQueue)

工作消息队列(WorkQueue)

 发布、订阅(Publish、Subscribe)

 Fanout Exchange

Direct Exchange 

Topic Exchange 

SpringAMQP-消息转换器 


一、同步通信 VS 异步通信

同步通信:双方在同一个时钟信号的控制下,进行数据的接收和发送,来一个时钟,发送端发送,接收端接收,他们彼此之间的工作状态是一致的,例如直播、打电话。

优点:

  • 时效性强,能够立即得到结果

缺点:

  • 耦合性较高:每次加入新的需求,都需要修改原有代码
  • 性能下降:调用者需要等待服务提供者响应,若调用链过长则响应时间等于每次调用时间之和
  • 资源利用率低:调用链中的每个服务在等待响应的过程中,不能释放请求占用的资源,高并发的情况下会造成资源的极度浪费
  • 级联失败:如果服务提供者出现问题,所有的调用方也会跟着出问题

适用场景:业务要求时效性高

异步通信:异步通信在发送字符时,所发送的字符之间的时间间隔可以是任意的。例如微信聊天。

在异步调用过程常见的实现就是事件驱动模式,系统中发生的事件会触发相应的事件处理器或监听器
,从而实现特定的业务逻辑或功能。

例如在如下的支付场景中,当有请求发送给支付服务时,支付服务就会通知Broker,接着后续的订阅事件就会接收到请求,开始同时处理业务,但是支付服务不用等到后续订阅事件完成后再返回,而是将请求通知给Broker之后支付服务就会返回结果。

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

优点:

  • 服务解耦
  • 性能提升,吞吐量提高
  • 服务之间没有强依赖,不用担心级联失败问题(故障隔离)
  • 流量削峰

缺点:

  • 依赖于Broker的可靠性、安全性和吞吐能力
  • 结构复杂后,业务没有了明显的流水线,难以追踪管理

适用场景:对于并发和吞吐量的要求高,时效性的要求低

二、MQ——消息队列

MQ(消息队列):存放消息的队列,也是事件驱动架构的Broker。

常见的消息队列实现对比:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

RabbitMQ 

RabbitMQ是基于Erlang语言开发的消息通信中间件,RabbitMQ的性能以及可用性较好,国内应用较为广泛,所以对RabbitMQ进行重点学习。

RabbitMQ的官网地址:https://www.rabbitmq.com

RabbitMQ安装 

可以根据自己的需求在RabbitMQ的官网进行查看:下载和安装 RabbitMQ — 兔子MQ

RabbitMQ的整体架构

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式​ 首先,Publisher会把消息发送给exchange(交换机),exchange负责路由再把消息投递到queue(队列),queue负责暂存消息,Consumer会从队列中获取消息并处理消息。

RabbitMQ中的几个概念:

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

常见消息模型 

RabbitMQ的官方文档中给出了5个MQ的Demo实例,可以分为如下:

  • 基本消息队列(BasicQueue)
  • 工作消息队列(WorkQueue)
  • 发布订阅(Publish、Subscribe),又根据交换机类型不同分为三种:

                Fanout Exchange:广播

                Direct Exchange:路由

                Topic Exchange:主题

 基本消息队列(BasicQueue)

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

  • publisher:消息发布者,将消息发送到队列queue
  • queue:消息队列,负责接受并缓存消息
  • consumer:订阅队列,处理队列中的消息
服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式
 
在RabbitMQ中需要了解的端口:
服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

在使用端口时,需要在云服务器上开放所用的端口 

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

  1. 建立Connection
  2. 创建Channel
  3. 利用Channel声明队列
  4. 利用Channel向队列中发送消息

代码实现:

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
        factory.setHost("x.x.x.x");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("xx");
        factory.setPassword("xx");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功:【" + message + "】");

        // 5.关闭通道和连接
        channel.close();
        connection.close();

    }
}

运行结果:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

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

  1. 建立Connection
  2. 创建Channel
  3. 利用Channel声明队列
  4. 定义Consumer的消费行为handleDelivery()
  5. 利用Channel将消费者与队列进行绑定

代码实现:

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数,分别是:主机名、端口号、vhost、用户名、密码
        factory.setHost("x.x.x.x");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("xx");
        factory.setPassword("xx");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息:【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}

运行结果:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

上述实现方式相对比较复杂,就引入了SpringAMQP来实现。

AMQP:是用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关,更符合微服务中独立性的要求。 

SpringAMQP:SpringAMQP是基于AMQP协议定义的一套API规范,提供了模板来发送和接收消息。包含两部分,其中spring-amqp是基础抽象,spring-rabbit是底层的默认实现。

SpringAMQP的官方地址

那么利用SpringAMQP来实现基本消息队列的流程如下:

  1. 在父工程中引入spring-amqp的依赖
  2. 在publisher服务中利用RabbitTemplate发送消息到simple.queue这个队列
  3. 在consumer服务中编写消费逻辑,绑定simple.queue这个队列

具体实现:

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

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

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

在publisher服务的配置文件中添加mq的连接信息:

spring:
  rabbitmq:
    host:  # rabbitMQ的ip地址
    port: 5672 # 端口
    username: # 用户名
    password: # 密码
    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);
    }
}

在RabbitMQ中的simple队列中查询信息:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

3、在consumer服务中编写消费逻辑,监听simple.queue

在consumer服务的配置文件中添加mq连接信息:

spring:
  rabbitmq:
    host:  # rabbitMQ的ip地址
    port: 5672 # 端口
    username: # 用户名
    password: # 密码
    virtual-host: # 虚拟主机

在consumer服务中新建一个类,编写具体的消费逻辑:

@Component
public class SpringRabbitListener { 
   @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueue(String msg) throws InterruptedException {
        System.out.println("消费者接收到消息:【" + msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }
}

运行启动类:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

工作消息队列(WorkQueue)

下面场景中如果queue中有50条请求消息,但是consumer1只能处理40条,剩余的10条就可以由consumer进行处理,所以说工作消息队列可以提高消息的处理速度,避免队列消息堆积

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

模拟Workqueue,实现一个队列绑定多个消费者,基本实现思路如下:

  1. 在publisher服务中定义测试方法,每秒产生50条消息,发送到simple.queue中
  2. 在consumer服务中定义两个消息监听者,都监听simple.queue队列
  3. 消费者1每秒处理50条消息,消费者2每秒处理10条消息

代码实现:

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

    public void testSendMessage2WorkQueue() throws InterruptedException {
        String queueName = "simple.queue";
        String message = "hello, message__";
        for (int i = 1; i <= 50; i++) {
            rabbitTemplate.convertAndSend(queueName, message + i);
            Thread.sleep(20);
        }
    }
在consumer服务中定义两个消息监听者,都监听simple.queue队列,设置消费者1每秒处理50条消息,消费者2每秒处理10条消息
    @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);
    }

修改application.yml文件,设置preFetch这个值,可以控制预取消息的上限,确保消费者2取消息时只能取一条,提高效率(“能者多劳”):

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 1

运行结果:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

 发布、订阅(Publish、Subscribe)

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

常见exchange类型包括:

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

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

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

 Fanout Exchange

 Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue中,如下:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

基本实现思路如下:

  1. 在consumer中,利用代码声明队列、交换机,将二者进行绑定
  2. 在consumer中,编写两个消费方法,分别监听fanout.queue1和fanout.queue2
  3. 在publisher中编写测试方法,向fanout发送消息

代码实现:

在consumer中,利用代码声明队列、交换机,将二者进行绑定

@Configuration
public class FanoutConfig {
    // itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.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);
    }
}

在consumer中,编写两个消费方法,分别监听fanout.queue1和fanout.queue2

    @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 + "】");
    }

在publisher中编写测试方法,向fanout发送消息

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

运行结果:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

Direct Exchange 

Direct Exchange会将接收到的消息根据规则路由到指定的Queue,因此被称为路由模式

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

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

基本实现思路如下:

  1. 利用@RabbitListener声明ExchangeQueueRoutingKey
  2. consumer服务中,编写两个消费者方法,分别监听direct.queue1direct.queue2
  3. publisher中编写测试方法,向itcast. direct发送消息

代码实现:

在consumer服务中,编写两个消费者方法,分别监听direct.queue1direct.queue2,并利用@RabbitListener声明ExchangeQueueRoutingKey

    @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服务发送消息到DirectExchange

    @Test
    public void testSendDirectExchange() {
        // 交换机名称
        String exchangeName = "itcast.direct";
        // 消息
        String message = "hello, red!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "red", message);
    }

运行结果:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

Topic Exchange 

Topic Exchange与Direct Exchange类似,区别在于Topic Exchange的routingKey必须是多个单词的列表,并且以.分割

QueueExchange指定BindingKey时可以使用通配符:

#:代指0个或多个单词

*:代指一个单词

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

基本实现思路如下:

  1. 利用@RabbitListener声明ExchangeQueueRoutingKey
  2. 在consumer服务中,编写两个消费者方法,分别监听topic.queue1topic.queue2
  3. 在publisher中编写测试方法,向itcast. topic发送消息

代码实现:

利用@RabbitListener声明ExchangeQueueRoutingKey,在consumer服务中,编写两个消费者方法,分别监听topic.queue1topic.queue2

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

在publisher中编写测试方法,向itcast. topic发送消息

    @Test
    public void testSendTopicExchange() {
        // 交换机名称
        String exchangeName = "itcast.topic";
        // 消息
        String message = "今天天气不错,我的心情好极了!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "china.weather", message);
    }

运行结果:

服务器的异步通信——RabbitMQ,微服务,rabbitmq,分布式

SpringAMQP-消息转换器 

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

Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter,基于JDKObjectOutputStream完成序列化。

如果要修改只需要定义一个MessageConverter 类型的Bean即可。

推荐用JSON方式序列化,实现步骤如下:

在父工程中引入依赖

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

在publisher和consumer服务中声明MessageConverter:文章来源地址https://www.toymoban.com/news/detail-820917.html

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

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

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

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

相关文章

  • 【学习日记2023.6.19】 之 RabbitMQ服务异步通信_消息可靠性_死信交换机_惰性队列_MQ集群

    消息队列在使用过程中,面临着很多实际问题需要思考: 消息从发送,到消费者接收,会经历多个过程: 其中的每一步都可能导致消息丢失,常见的丢失原因包括: 发送时丢失: 生产者发送的消息未送达exchange 消息到达exchange后未到达queue MQ宕机,queue将消息丢失 consumer接收

    2024年02月11日
    浏览(53)
  • 探索 XMLHttpRequest:网页与服务器的异步通信之道(下)

    🤍 前端开发工程师、技术日更博主、已过CET6 🍨 阿珊和她的猫_ CSDN 博客专家、23年度博客之星前端领域TOP1 🕠 牛客 高级专题作者、打造专栏《前端面试必备》 、《2024面试高频手撕题》 🍚 蓝桥云课 签约作者、上架课程《Vue.js 和 Egg.js 开发企业级健康管理项目》、《带你

    2024年02月20日
    浏览(42)
  • 探索 XMLHttpRequest:网页与服务器的异步通信之道(上)

    🤍 前端开发工程师、技术日更博主、已过CET6 🍨 阿珊和她的猫_ CSDN 博客专家、23年度博客之星前端领域TOP1 🕠 牛客 高级专题作者、打造专栏《前端面试必备》 、《2024面试高频手撕题》 🍚 蓝桥云课 签约作者、上架课程《Vue.js 和 Egg.js 开发企业级健康管理项目》、《带你

    2024年02月21日
    浏览(42)
  • Windows 下安装RabbitMQ服务器(超详细)

            RabbitMQ是一个在AMQP协议标准基础上完整的,可复用的企业消息系统。它遵循Mozilla Public License开源协议,采用 Erlang 实现的工业级的消息队列(MQ)服务器,Rabbit MQ 是建立在Erlang OTP平台上。         安装RabbitMQ服务器必须首先安装 Erlang 运行环境。         安装

    2024年04月14日
    浏览(35)
  • Docker (CentOS)配置rabbitMQ 服务器访问15672端口显示服务器拒绝连接

    在自己的服务器上配置rabbitMq时已经启动成功了,但是访问就是出错。 使用以下命令启动。  启动后可以看到已经启动成功。 输入地址去访问看到被拒绝访问了  那么有三种可能性:1、你的防火墙端口没有开放 使用命令开放端口  2、你的服务器安全组没有添加该端口,去你

    2024年02月14日
    浏览(42)
  • 使用Docker安装运行RabbitMQ---阿里云服务器

    目录 0、阿里云没开端口的得要去安全组规则去添加: 1、下载RabbitMQ镜像: 2、查看镜像是否下载成功,得到docker镜像id: 3、运行RabbitMQ: 4、查看RabbbitMQ容器是否启动成功: 5、启动RabbitMQ中的插件管理 6、访问RabbitMQ的web配置界面 7、运行成功 0、阿里云没开端口的得要去安全

    2024年03月12日
    浏览(57)
  • 三台Windows服务器搭建RabbitMq集群及镜像队列

    提示:文章写完后,目录可以自动生成,如何生成可参考右边的帮助文档 消息中间件-RabbitMq集群搭建及镜像队列 (高可用性、一台宕机自动切换另一台) ,镜像队列是基于普通的集群模式的 提示:以下是本篇文章正文内容,下面案例可供参考 RabbitMQ是实现了高级消息队列协议

    2024年02月01日
    浏览(49)
  • Spring Boot 项目应用消息服务器RabbitMQ(简单介绍)

    本章讲述的是在用户下单环节,消息服务器RabbitMQ 的应用 在写一个电商项目的小demo,在电商项目中,消息服务器的应用: 1、订单状态通知:当用户下单、支付成功、订单发货、订单完成等关键节点时,可以通过消息服务器向用户发送相应的订单状态通知。 2、消息推送:通

    2024年02月13日
    浏览(79)
  • 【软件环境安装部署】华为云服务器 Docker安装 RabbitMQ

    RabbitMQ简称MQ是一套实现了高级消息队列协议的开源消息代理软件,简单来说就是一个消息中间件。是一种 程序对程序的通信方法,其服务器也是以高性能、健壮以及可伸缩性出名的Erlang语言编写而成。 RabbitMQ简单来说就是一个消息队列中间件,用来保存消息和传递消息的一

    2024年02月13日
    浏览(46)
  • 如何才能在Ubuntu系统部署RabbitMQ服务器并公网访问

    在Ubuntu系统上部署RabbitMQ服务器并公网访问,可以按照以下步骤进行: 安装RabbitMQ服务器: 在终端中输入以下命令安装RabbitMQ服务器: 启动RabbitMQ服务器: 在终端中输入以下命令启动RabbitMQ服务器: 设置RabbitMQ服务器开机启动: 在终端中输入以下命令设置RabbitMQ服务器开机启

    2024年02月07日
    浏览(48)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包