精华内容
下载资源
问答
  • 消息队列原理

    2019-09-09 11:29:57
  • 什么是消息队列及消息队列原理和应用场景详解 转载 ...

    什么是消息队列及消息队列原理和应用场景详解

    “消息队列”是在消息的传输过程中保存消息的容器。
    “消息”是在两台计算机间传送的数据单位。消息可以非常简单,例如只包含文本字符串;也可以更复杂,可能包含嵌入对象。
    消息被发送到队列中。“消息队列”是在消息的传输过程中保存消息的容器。消息队列管理器在将消息从它的源中继到它的目标时充当中间人。队列的主要目的是提供路由并保证消息的传递;如果发送消息时接收者不可用,消息队列会保留消息,直到可以成功地传递它。

    为什么会需要消息队列(MQ)?

    主要原因是由于在高并发环境下,由于来不及同步处理,请求往往会发生堵塞,比如说,大量的insert,update之类的请求同时到达MySQL,直接导致无数的行锁表锁,甚至最后请求会堆积过多,从而触发too many connections错误。通过使用消息队列,我们可以异步处理请求,从而缓解系统的压力。

    美国计算机科学家,LaTex的作者Leslie Lamport说:“分布式系统就是这样一个系统,系统中一个你甚至都不知道的计算机出了故障,却可能导致你自己的计算机不可用。”一语道破了开发分布式系统的玄机,那就是它的复杂与不可控。所以Martin Fowler强调:分布式调用的第一原则就是不要分布式。这句话看似颇具哲理,然而就企业应用系统而言,只要整个系统在不停地演化,并有多个子系统共同存在时,这条原则就会被迫打破。盖因为在当今的企业应用系统中,很难寻找到完全不需要分布式调用的场景。Martin Fowler提出的这条原则,一方面是希望设计者能够审慎地对待分布式调用,另一方面却也是分布式系统自身存在的缺陷所致。无论是CORBA,还是EJB 2;无论是RPC平台,还是Web Service,都因为驻留在不同进程空间的分布式组件,而引入额外的复杂度,并可能对系统的效率、可靠性、可预测性等诸多方面带来负面的影响。

    然而,不可否认的是在企业应用系统领域,我们总是会面对不同系统之间的通信、集成与整合,尤其当面临异构系统时,这种分布式的调用与通信变得越重要,它在架构设计中就更加凸显其价值。并且,从业务分析与架构质量的角度来讲,我们也希望在系统架构中尽可能地形成对服务的重用,通过独立运行在进程中服务的形式,彻底解除客户端与服务端的耦合。这常常是架构演化的必然道路。在我的同事陈金洲发表在InfoQ上的文章《架构腐化之谜》中,就认为可以通过“将独立的模块放入独立的进程”来解决架构因为代码规模变大而腐化的问题。

    随着网络基础设施的逐步成熟,从RPC进化到Web Service,并在业界开始普遍推行SOA,再到后来的RESTful平台以及云计算中的PaaS与SaaS概念的推广,分布式架构在企业应用中开始呈现出不同的风貌,然而殊途同归,这些分布式架构的目标仍然是希望回到建造巴别塔的时代,系统之间的交流不再为不同语言与平台的隔阂而产生障碍。正如Martin Fowler在《企业集成模式》一书的序中写道:“集成之所以重要是因为相互独立的应用是没有生命力的。我们需要一种技术能将在设计时并未考虑互操作的应用集成起来,打破它们之间的隔阂,获得比单个应用更多的效益”。这或许是分布式架构存在的主要意义。
     

    目录

    1. 消息队列概述
    2. 消息队列应用场景
    3. 消息中间件示例
    4. JMS消息服务
    5. 常用消息队列
    6. 参考(推荐)资料
    7. 本次分享总结

    一、消息队列概述

    消息队列中间件是分布式系统中重要的组件,主要解决应用耦合,异步消息,流量削锋等问题。实现高性能,高可用,可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。

    目前在生产环境,使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,RocketMQ等。

    二、消息队列应用场景

    以下介绍消息队列在实际应用中常用的使用场景。异步处理,应用解耦,流量削锋和消息通讯四个场景。

    2.1异步处理

    场景说明:用户注册后,需要发注册邮件和注册短信。传统的做法有两种1.串行的方式;2.并行方式。

    (1)串行方式:将注册信息写入数据库成功后,发送注册邮件,再发送注册短信。以上三个任务全部完成后,返回给客户端。

     

    (2)并行方式:将注册信息写入数据库成功后,发送注册邮件的同时,发送注册短信。以上三个任务完成后,返回给客户端。与串行的差别是,并行的方式可以提高处理的时间。

     

    假设三个业务节点每个使用50毫秒钟,不考虑网络等其他开销,则串行方式的时间是150毫秒,并行的时间可能是100毫秒。

    因为CPU在单位时间内处理的请求数是一定的,假设CPU1秒内吞吐量是100次。则串行方式1秒内CPU可处理的请求量是7次(1000/150)。并行方式处理的请求量是10次(1000/100)。

     

    小结:如以上案例描述,传统的方式系统的性能(并发量,吞吐量,响应时间)会有瓶颈。如何解决这个问题呢?

    引入消息队列,将不是必须的业务逻辑,异步处理。改造后的架构如下:

     

    按照以上约定,用户的响应时间相当于是注册信息写入数据库的时间,也就是50毫秒。注册邮件,发送短信写入消息队列后,直接返回,因此写入消息队列的速度很快,基本可以忽略,因此用户的响应时间可能是50毫秒。因此架构改变后,系统的吞吐量提高到每秒20 QPS。比串行提高了3倍,比并行提高了两倍。

    2.2应用解耦

    场景说明:用户下单后,订单系统需要通知库存系统。传统的做法是,订单系统调用库存系统的接口。如下图:

     

    传统模式的缺点:

    1)  假如库存系统无法访问,则订单减库存将失败,从而导致订单失败;

    2)  订单系统与库存系统耦合;

    如何解决以上问题呢?引入应用消息队列后的方案,如下图:

     

    • 订单系统:用户下单后,订单系统完成持久化处理,将消息写入消息队列,返回用户订单下单成功。
    • 库存系统:订阅下单的消息,采用拉/推的方式,获取下单信息,库存系统根据下单信息,进行库存操作。
    • 假如:在下单时库存系统不能正常使用。也不影响正常下单,因为下单后,订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦。

    2.3流量削锋

    流量削锋也是消息队列中的常用场景,一般在秒杀或团抢活动中使用广泛。

    应用场景:秒杀活动,一般会因为流量过大,导致流量暴增,应用挂掉。为解决这个问题,一般需要在应用前端加入消息队列。

    1. 可以控制活动的人数;
    2. 可以缓解短时间内高流量压垮应用;

     

    1. 用户的请求,服务器接收后,首先写入消息队列。假如消息队列长度超过最大数量,则直接抛弃用户请求或跳转到错误页面;
    2. 秒杀业务根据消息队列中的请求信息,再做后续处理。

    2.4日志处理

    日志处理是指将消息队列用在日志处理中,比如Kafka的应用,解决大量日志传输的问题。架构简化如下:

     

    • 日志采集客户端,负责日志数据采集,定时写受写入Kafka队列;
    • Kafka消息队列,负责日志数据的接收,存储和转发;
    • 日志处理应用:订阅并消费kafka队列中的日志数据;


    以下是新浪kafka日志处理应用案例:

     

    (1)Kafka:接收用户日志的消息队列。

    (2)Logstash:做日志解析,统一成JSON输出给Elasticsearch。

    (3)Elasticsearch:实时日志分析服务的核心技术,一个Schemaless,实时的数据存储服务,通过index组织数据,兼具强大的搜索和统计功能。

    (4)Kibana:基于Elasticsearch的数据可视化组件,超强的数据可视化能力是众多公司选择ELK stack的重要原因。

    2.5消息通讯

    消息通讯是指,消息队列一般都内置了高效的通信机制,因此也可以用在纯的消息通讯。比如实现点对点消息队列,或者聊天室等。

    点对点通讯:

     

    客户端A和客户端B使用同一队列,进行消息通讯。

    聊天室通讯:

     

    客户端A,客户端B,客户端N订阅同一主题,进行消息发布和接收。实现类似聊天室效果。

    以上实际是消息队列的两种消息模式,点对点或发布订阅模式。模型为示意图,供参考。

    三、消息中间件示例

    3.1电商系统

     

    消息队列采用高可用,可持久化的消息中间件。比如Active MQ,Rabbit MQ,Rocket Mq。(1)应用将主干逻辑处理完成后,写入消息队列。消息发送是否成功可以开启消息的确认模式。(消息队列返回消息接收成功状态后,应用再返回,这样保障消息的完整性)

    (2)扩展流程(发短信,配送处理)订阅队列消息。采用推或拉的方式获取消息并处理。

    (3)消息将应用解耦的同时,带来了数据一致性问题,可以采用最终一致性方式解决。比如主数据写入数据库,扩展应用根据消息队列,并结合数据库方式实现基于消息队列的后续处理。

    3.2日志收集系统

     

    分为Zookeeper注册中心,日志收集客户端,Kafka集群和Storm集群(OtherApp)四部分组成。

    • Zookeeper注册中心,提出负载均衡和地址查找服务;
    • 日志收集客户端,用于采集应用系统的日志,并将数据推送到kafka队列;
    • Kafka集群:接收,路由,存储,转发等消息处理;

    Storm集群:与OtherApp处于同一级别,采用拉的方式消费队列中的数据;

    四、JMS消息服务

    讲消息队列就不得不提JMS 。JMS(JAVA Message Service,java消息服务)API是一个消息服务的标准/规范,允许应用程序组件基于JavaEE平台创建、发送、接收和读取消息。它使分布式通信耦合度更低,消息服务更加可靠以及异步性。

    在EJB架构中,有消息bean可以无缝的与JM消息服务集成。在J2EE架构模式中,有消息服务者模式,用于实现消息与应用直接的解耦。

    4.1消息模型

    在JMS标准中,有两种消息模型P2P(Point to Point),Publish/Subscribe(Pub/Sub)。

    4.1.1 P2P模式

     

    P2P模式包含三个角色:消息队列(Queue),发送者(Sender),接收者(Receiver)。每个消息都被发送到一个特定的队列,接收者从队列中获取消息。队列保留着消息,直到他们被消费或超时。

     

    P2P的特点

    • 每个消息只有一个消费者(Consumer)(即一旦被消费,消息就不再在消息队列中)
    • 发送者和接收者之间在时间上没有依赖性,也就是说当发送者发送了消息之后,不管接收者有没有正在运行,它不会影响到消息被发送到队列
    • 接收者在成功接收消息之后需向队列应答成功

     

    如果希望发送的每个消息都会被成功处理的话,那么需要P2P模式。

    4.1.2 Pub/sub模式

     

    包含三个角色主题(Topic),发布者(Publisher),订阅者(Subscriber) 。多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。

    Pub/Sub的特点

    • 每个消息可以有多个消费者
    • 发布者和订阅者之间有时间上的依赖性。针对某个主题(Topic)的订阅者,它必须创建一个订阅者之后,才能消费发布者的消息。
    • 为了消费消息,订阅者必须保持运行的状态。

     

    为了缓和这样严格的时间相关性,JMS允许订阅者创建一个可持久化的订阅。这样,即使订阅者没有被激活(运行),它也能接收到发布者的消息。

    如果希望发送的消息可以不被做任何处理、或者只被一个消息者处理、或者可以被多个消费者处理的话,那么可以采用Pub/Sub模型。

    4.2消息消费

    在JMS中,消息的产生和消费都是异步的。对于消费来说,JMS的消息者可以通过两种方式来消费消息。

    (1)同步

    订阅者或接收者通过receive方法来接收消息,receive方法在接收到消息之前(或超时之前)将一直阻塞;

    (2)异步

    订阅者或接收者可以注册为一个消息监听器。当消息到达之后,系统自动调用监听器的onMessage方法。

     

    JNDI:Java命名和目录接口,是一种标准的Java命名系统接口。可以在网络上查找和访问服务。通过指定一个资源名称,该名称对应于数据库或命名服务中的一个记录,同时返回资源连接建立所必须的信息。

    JNDI在JMS中起到查找和访问发送目标或消息来源的作用。

    4.3JMS编程模型

    (1) ConnectionFactory

    创建Connection对象的工厂,针对两种不同的jms消息模型,分别有QueueConnectionFactory和TopicConnectionFactory两种。可以通过JNDI来查找ConnectionFactory对象。

    (2) Destination

    Destination的意思是消息生产者的消息发送目标或者说消息消费者的消息来源。对于消息生产者来说,它的Destination是某个队列(Queue)或某个主题(Topic);对于消息消费者来说,它的Destination也是某个队列或主题(即消息来源)。

    所以,Destination实际上就是两种类型的对象:Queue、Topic可以通过JNDI来查找Destination。

    (3) Connection

    Connection表示在客户端和JMS系统之间建立的链接(对TCP/IP socket的包装)。Connection可以产生一个或多个Session。跟ConnectionFactory一样,Connection也有两种类型:QueueConnection和TopicConnection。

    (4) Session

    Session是操作消息的接口。可以通过session创建生产者、消费者、消息等。Session提供了事务的功能。当需要使用session发送/接收多个消息时,可以将这些发送/接收动作放到一个事务中。同样,也分QueueSession和TopicSession。

    (5) 消息的生产者

    消息生产者由Session创建,并用于将消息发送到Destination。同样,消息生产者分两种类型:QueueSender和TopicPublisher。可以调用消息生产者的方法(send或publish方法)发送消息。

    (6) 消息消费者

    消息消费者由Session创建,用于接收被发送到Destination的消息。两种类型:QueueReceiver和TopicSubscriber。可分别通过session的createReceiver(Queue)或createSubscriber(Topic)来创建。当然,也可以session的creatDurableSubscriber方法来创建持久化的订阅者。

    (7) MessageListener

    消息监听器。如果注册了消息监听器,一旦消息到达,将自动调用监听器的onMessage方法。EJB中的MDB(Message-Driven Bean)就是一种MessageListener。

     

    深入学习JMS对掌握JAVA架构,EJB架构有很好的帮助,消息中间件也是大型分布式系统必须的组件。本次分享主要做全局性介绍,具体的深入需要大家学习,实践,总结,领会。

    五、常用消息队列

    一般商用的容器,比如WebLogic,JBoss,都支持JMS标准,开发上很方便。但免费的比如Tomcat,Jetty等则需要使用第三方的消息中间件。本部分内容介绍常用的消息中间件(Active MQ,Rabbit MQ,Zero MQ,Kafka)以及他们的特点。

    5.1 ActiveMQ

    ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

    ActiveMQ特性如下:

    ⒈ 多种语言和协议编写客户端。语言: Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP

    ⒉ 完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)

    ⒊ 对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性

    ⒋ 通过了常见J2EE服务器(如 Geronimo,JBoss 4,GlassFish,WebLogic)的测试,其中通过JCA 1.5 resource adaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上

    ⒌ 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA

    ⒍ 支持通过JDBC和journal提供高速的消息持久化

    ⒎ 从设计上保证了高性能的集群,客户端-服务器,点对点

    ⒏ 支持ajax

    ⒐ 支持与Axis的整合

    ⒑ 可以很容易得调用内嵌JMS provider,进行测试

    5.2 RabbitMQ

    RabbitMQ是流行的开源消息队列系统,用erlang语言开发。RabbitMQ是AMQP(高级消息队列协议)的标准实现。支持多种客户端,如:Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等,支持AJAX,持久化。用于在分布式系统中存储转发消息,在易用性、扩展性、高可用性等方面表现不俗。

    结构图如下:

    几个重要概念:

    Broker:简单来说就是消息队列服务器实体。

      Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列。

      Queue:消息队列载体,每个消息都会被投入到一个或多个队列。

      Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来。

      Routing Key:路由关键字,exchange根据这个关键字进行消息投递。

      vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离。

      producer:消息生产者,就是投递消息的程序。

      consumer:消息消费者,就是接受消息的程序。

      channel:消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务。

    消息队列的使用过程,如下:

    (1)客户端连接到消息队列服务器,打开一个channel。

    (2)客户端声明一个exchange,并设置相关属性。

    (3)客户端声明一个queue,并设置相关属性。

    (4)客户端使用routing key,在exchange和queue之间建立好绑定关系。

    (5)客户端投递消息到exchange。

    exchange接收到消息后,就根据消息的key和已经设置的binding,进行消息路由,将消息投递到一个或多个队列里。

    5.3 ZeroMQ

    号称史上最快的消息队列,它实际类似于Socket的一系列接口,他跟Socket的区别是:普通的socket是端到端的(1:1的关系),而ZMQ却是可以N:M 的关系,人们对BSD套接字的了解较多的是点对点的连接,点对点连接需要显式地建立连接、销毁连接、选择协议(TCP/UDP)和处理错误等,而ZMQ屏蔽了这些细节,让你的网络编程更为简单。ZMQ用于node与node间的通信,node可以是主机或者是进程。

    引用官方的说法: “ZMQ(以下ZeroMQ简称ZMQ)是一个简单好用的传输层,像框架一样的一个socket library,他使得Socket编程更加简单、简洁和性能更高。是一个消息处理队列库,可在多个线程、内核和主机盒之间弹性伸缩。ZMQ的明确目标是“成为标准网络协议栈的一部分,之后进入Linux内核”。现在还未看到它们的成功。但是,它无疑是极具前景的、并且是人们更加需要的“传统”BSD套接字之上的一 层封装。ZMQ让编写高性能网络应用程序极为简单和有趣。”

    特点是:

    • 高性能,非持久化;
    • 跨平台:支持Linux、Windows、OS X等。
    • 多语言支持; C、C++、Java、.NET、Python等30多种开发语言。
    • 可单独部署或集成到应用中使用;
    • 可作为Socket通信库使用。

    与RabbitMQ相比,ZMQ并不像是一个传统意义上的消息队列服务器,事实上,它也根本不是一个服务器,更像一个底层的网络通讯库,在Socket API之上做了一层封装,将网络通讯、进程通讯和线程通讯抽象为统一的API接口。支持“Request-Reply “,”Publisher-Subscriber“,”Parallel Pipeline”三种基本模型和扩展模型。

     

    ZeroMQ高性能设计要点:

    1、无锁的队列模型

       对于跨线程间的交互(用户端和session)之间的数据交换通道pipe,采用无锁的队列算法CAS;在pipe两端注册有异步事件,在读或者写消息到pipe的时,会自动触发读写事件。

    2、批量处理的算法

       对于传统的消息处理,每个消息在发送和接收的时候,都需要系统的调用,这样对于大量的消息,系统的开销比较大,zeroMQ对于批量的消息,进行了适应性的优化,可以批量的接收和发送消息。

    3、多核下的线程绑定,无须CPU切换

       区别于传统的多线程并发模式,信号量或者临界区, zeroMQ充分利用多核的优势,每个核绑定运行一个工作者线程,避免多线程之间的CPU切换开销。

    5.4 Kafka

    Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据。 这种动作(网页浏览,搜索和其他用户的行动)是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop的一样的日志数据和离线分析系统,但又要求实时处理的限制,这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理,也是为了通过集群机来提供实时的消费。

    Kafka是一种高吞吐量的分布式发布订阅消息系统,有如下特性:

    • 通过O(1)的磁盘数据结构提供消息的持久化,这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。(文件追加的方式写入数据,过期的数据定期删除)
    • 高吞吐量:即使是非常普通的硬件Kafka也可以支持每秒数百万的消息。
    • 支持通过Kafka服务器和消费机集群来分区消息。
    • 支持Hadoop并行数据加载。

     

    Kafka相关概念

    • Broker

    Kafka集群包含一个或多个服务器,这种服务器被称为broker[5]

    • Topic

    每条发布到Kafka集群的消息都有一个类别,这个类别被称为Topic。(物理上不同Topic的消息分开存储,逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处)

    • Partition

    Parition是物理上的概念,每个Topic包含一个或多个Partition.

    • Producer

    负责发布消息到Kafka broker

    • Consumer

    消息消费者,向Kafka broker读取消息的客户端。

    • Consumer Group

    每个Consumer属于一个特定的Consumer Group(可为每个Consumer指定group name,若不指定group name则属于默认的group)。

     

    一般应用在大数据日志处理或对实时性(少量延迟),可靠性(少量丢数据)要求稍低的场景使用。

    六、参考资料

    以下是本次分享参考的资料和推荐大家参考的资料。

     

    参考资料(可参考资料):

    (1)Jms

    http://blog.sina.com.cn/s/blog_3fba24680100r777.html

    http://blog.csdn.net/jiuqiyuliang/article/details/46701559(深入浅出JMS(一)--JMS基本概念)

    (2)RabbitMQ

    http://baike.baidu.com/link?url=s2cU-QgOsXan7j0AM5qxxlmruz6WEeBQXX-Bbk0O3F5jt9Qts2uYQARxQxl7CBT2SO2NF2VkzX_XZLqU-CTaPa

    http://blog.csdn.net/sun305355024sun/article/details/41913105

    (3)Zero MQ

    http://www.searchtb.com/2012/08/zeromq-primer.html

    http://blog.csdn.net/yangbutao/article/details/8498790

    http://wenku.baidu.com/link?url=yYoiZ_pYPCuUxEsGQvMMleY08bcptZvwF3IMHo2W1i-ti66YXXPpLLJBGXboddwgGBnOehHiUdslFhtz7RGZYkrtMQQ02DV5sv9JFF4LZnK

    (4)Kafka

    http://baike.baidu.com/link?url=qQXyqvPQ1MVrw9WkOGSGEfSX1NHy4unsgc4ezzJwU94SrPuVnrKf2tbm4SllVaN3ArGGxV_N5hw8JTT2-lw4QK

    http://www.infoq.com/cn/articles/apache-kafka/

    http://www.mincoder.com/article/3942.shtml

     

    七、本次分享总结

    展开全文
  • rabbitmq消息队列原理

    2020-09-26 10:33:07
    转载:rabbitmq消息队列原理 一、rabbitmq架构 RabbitMQ是一个流行的开源消息队列系统,是AMQP(高级消息队列协议)标准的实现,由以高性能、健壮、可伸缩性出名的Erlang语言开发,并继承了这些优点。...

    转载:rabbitmq消息队列原理

    一、rabbitmq架构

    RabbitMQ是一个流行的开源消息队列系统,是AMQP(高级消息队列协议)标准的实现,由以高性能、健壮、可伸缩性出名的Erlang语言开发,并继承了这些优点。rabbitmq简单架构如下:
    这里写图片描述
    上图简单展示了rabbitmq的架构,从图中看到几个关键字:vhost、exchange、route key、queue等,后面会介绍这些概念。

    下面看下rabbitmq的进程模型:
    这里写图片描述
    看到这个图,相信大家应该很熟悉,没错就是事件驱动模型(或者说反应堆模型),这是一种高性能的非阻塞io线程模型,不过在Erlang中称为进程模型。

    tcp_acceptor进程接收客户端连接,创建rabbit_reader、rabbit_writer、rabbit_channel进程。
    rabbit_reader接收客户端连接,解析AMQP帧;rabbit_writer向客户端返回数据;
    rabbit_channel解析AMQP方法,对消息进行路由,然后发给相应队列进程。
    rabbit_amqqueue_process是队列进程,在RabbitMQ启动(恢复durable类型队列)或创建队列时创建。
    rabbit_msg_store是负责消息持久化的进程。

    在整个系统中,存在一个tcp_accepter进程,一个rabbit_msg_store进程,有多少个队列就有多少个rabbit_amqqueue_process进程,每个客户端连接对应一个rabbit_reader和rabbit_writer进程。
    二、关于AMQP协议

    1.AMQP帧组件

    AMQP帧由五个不同的组件组成:

    帧类型
    信道编号
    以字节为单位的帧大小
    帧有效载荷payload
    结束字节标志(ASCII值206)
    
    • 1
    • 2
    • 3
    • 4
    • 5

    这里写图片描述
    2.帧类型

    AMQP规范定义了五种类型的帧:协议头帧、方法帧、内容帧、消息体帧及心跳帧。每种帧类型都有明确的目的,有些帧的使用频率比其他的高很多:

    协议头帧用于连接到rabbitmq,进使用一次。
    方法帧携带发送给rabbitmq或者从rabbitmq接收到的rpc请求或者响应
    内容头包含一条消息的大小和属性。
    消息体帧包含消息的内容
    心跳帧在客户端与rabbitmq直接进行传递,作为一种校验机制确保连接的两端都可用并且正常工作。
    
    • 1
    • 2
    • 3
    • 4
    • 5

    3.将消息编组成帧

    我们使用方法帧、内容头帧和消息体帧组成一个完整的rabbitmq消息。方法头帧携带命令和执行它所需要的参数(如交换器和路由键)、内容帧包含消息的基本属性以及消息的大小,消息体帧也就是携带我们真正需要发送的消息内容。
    这里写图片描述
    4.方法帧结构
    这里写图片描述
    5.内容头帧结构
    这里写图片描述
    内容头包含的具体属性如下:
    这里写图片描述
    content-type:消息体的报文编码,如application/json
    expiration:消息过期时间
    reply-to:响应消息的队列名
    content-encoding:报文压缩的编码,如gzip
    message-id:消息的编号
    correlation-id:链路id
    deliver-mode:告诉rabbitmq将消息写入磁盘还是内存
    user-id:投递消息的用户(发送消息时不要设置该值)
    timestamp:投递消息的时间
    headers:定义一些属性,可用于实现rabbitmq路由(比如exchange类型是headers的时候用到)

    6.消息体帧结构
    这里写图片描述
    7.几个概念:
    Broker:简单来说就是消息队列服务器实体
    Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列
    Queue:消息队列载体,每个消息都会被投入到一个或多个队列,队列类型又分为临时队列,持久化队列,排他队列
    Binding:绑定,它的作用就是把exchange和queue按照路由规则绑定起来
    Routing Key:路由关键字,exchange根据这个关键字进行消息投递
    vhost:虚拟主机,一个broker里可以开设多个vhost,用作不同用户的权限分离
    producer:消息生产者,就是投递消息的程序
    consumer:消息消费者,就是接受消息的程序
    channel:消息通道,在客户端的每个连接里,可建立多个channel,每个channel代表一个会话任务

    四、通讯过程
    1.启动会话
    这里写图片描述
    2.声明交换器
    这里写图片描述
    3.声明队列
    这里写图片描述
    4.绑定队列到exchange
    这里写图片描述
    5.发送消息-使用事务机制
    这里写图片描述
    对事务的支持是AMQP协议的一个重要特性。假设当生产者将一个持久化消息发送给服务器时,假如使用no_ack模式,所以即使服务器崩溃,没有持久化该消息,生产者也无法获知该消息已经丢失。如果此时使用事务,即通过txSelect()开启一个事务,然后发送消息给服务器,然后通过txCommit()提交该事务,即可以保证,如果txCommit()提交了,则该消息一定会持久化,如果txCommit()还未提交即服务器崩溃,则该消息不会服务器就收。当然Rabbit MQ也提供了txRollback()命令用于回滚某一个事务。但是使用事务,会导致性能下降,它使得生产者发布消息后必须等到消息真正持久化后服务端响应了才结束本次连接,所以需要在实际应用中平衡性能与安全的问题。

    6.发送消息-非事务方式
    这里写图片描述
    使用事务固然可以保证只有提交的事务,才会被服务器执行。但是这样同时也将客户端与消息服务器同步起来,这背离了消息队列解耦的本质。Rabbit MQ提供了一个更加轻量级的机制来保证生产者可以感知服务器消息是否已被路由到正确的队列中——Confirm。如果设置channel为confirm状态,则通过该channel发送的消息都会被分配一个唯一的ID,然后一旦该消息被正确的路由到匹配的队列中后,服务器会返回给生产者一个Confirm,该Confirm包含该消息的ID,这样生产者就会知道该消息已被正确分发。对于持久化消息,只有该消息被持久化后,才会返回Confirm。

    Confirm机制的最大优点在于异步,生产者在发送消息以后,即可继续执行其他任务(也就是异步监听服务端的ack即可)。而服务器返回Confirm后,会触发生产者的回调函数,生产者在回调函数中处理Confirm信息。如果消息服务器发生异常,导致该消息丢失,会返回给生产者一个nack,表示消息已经丢失,这样生产者就可以通过重发消息,保证消息不丢失。Confirm机制在性能上要比事务优越很多。

    但是Confirm机制,无法进行回滚,就是一旦服务器崩溃,生产者无法得到Confirm信息,生产者其实本身也不知道该消息吃否已经被持久化,只有继续重发来保证消息不丢失,但是如果原先已经持久化的消息,并不会被回滚,这样队列中就会存在两条相同的消息,系统需要支持去重。

    7.消费消息
    这里写图片描述

    五、使用delivery-mode平衡速度和安全
    delivery-mode有两个值:1表示非持久化,2表示持久化消息

    1.发送消息到纯内存队列中,delivery-mode = 1
    这里写图片描述
    特点:非持久化的消息在服务宕机的时候会丢失数据,但是由于不需要磁盘io,尽可能地降低消息投递的延迟性,性能较高。

    2.发布消息到支持磁盘存储的队列,delivery-mode = 2
    这里写图片描述
    特点:持久化的消息安全性较高,尽管服务宕机,数据也不会丢失,但是在投递消息的过程中需要发生磁盘io,性能相对纯内存投递的方式低,但是尽管是产生了磁盘io,由于日志的记录方式是直接追加到消息日志文件的末尾,属于顺序io,没有随机io,所以性能还是可以接受的。

    1. 大概原理:
      所有队列中的消息都以append的方式写到一个文件中,当这个文件的大小超过指定的限制大小后,关闭这个文件再创建一个新的文件供消息的写入。文件名(*.rdq)从0开始然后依次累加。当某个消息被删除时,并不立即从文件中删除相关信息,而是做一些记录,当垃圾数据达到一定比例时,启动垃圾回收处理,将逻辑相邻的文件中的数据合并到一个文件中。

    2. 消息的读写及删除:
      rabbitmq在启动时会创建msg_store_persistent,msg_store_transient两个进程,一个用于持久消息的存储,一个用于内存不够时,将存储在内存中的非持久化数据转存到磁盘中。所有队列的消息的写入和删除最终都由这两个进程负责处理,而消息的读取则可能是队列本身直接打开文件进行读取,也可能是发送请求由msg_store_persisteng/msg_store_transient进程进行处理。

    在进行消息的存储时,rabbitmq会在ets表中记录消息在文件中的映射,以及文件的相关信息。消息读取时,根据消息ID找到该消息所存储的文件,在文件中的偏移量,然后打开文件进行读取。消息的删除只是从ets表删除指定消息的相关信息,同时更新消息对应存储的文件的相关信息(更新文件有效数据大小)。
    六、消息路由模式
    1.fanout模式
    fanout类型的Exchange路由规则非常简单,它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。
    这里写图片描述
    上图中,生产者发送到Exchange的所有消息都会路由到图中的两个Queue,并最终被两个消费者(C1与C2)消费。
    2.direct模式
    direct类型的Exchange路由规则也很简单,它会把消息路由到那些binding key与routing key完全匹配的Queue中。如图,生产者发送消息的routing key=key1的时候,只有绑定了key1的queue才能收到信息
    这里写图片描述
    3.topic模式
    topic类型的Exchange在匹配规则上进行了扩展,它与direct类型的Exchage相似,也是将消息路由到binding key与routing key相匹配的Queue中,但这里的匹配规则有些不同,它约定:
    routing key为一个句点号“. ”分隔的字符串(我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词),如“image.new.profile”.
    binding key与routing key一样也是句点号“. ”分隔的字符串
    binding key中可以存在两种特殊字符“”与“#”,用于做模糊匹配,其中“”用于匹配下一个据点前的所有字符,“#”用于匹配所有字符,包括句点(可以是零个)
    这里写图片描述
    如图,生产者以routing key为image.new.profile发布消息,这key可以被image.*.profile以及image.#匹配到,所有这两个队列都可以收到消息。由此可见,topic的路由方式更加灵活。
    3.headers模式
    headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息,而是根据发送的消息内容中的headers属性进行匹配。
    在绑定Queue与Exchange时指定一组键值对以及x-match参数,x-match参数是字符串类型,可以设置为any或者all。如果设置为any,意思就是只要匹配到了headers表中的任何一对键值即可,all则代表需要全部匹配。

    七、rabbitmq流量控制

    RabbitMQ可以对内存和磁盘使用量设置阈值,当达到阈值后,生产者将被阻塞(block),直到对应项恢复正常。除了这两个阈值,RabbitMQ在正常情况下还用流控(Flow Control)机制来确保稳定性。
    Erlang进程之间并不共享内存(binaries类型除外),而是通过消息传递来通信,每个进程都有自己的进程邮箱。Erlang默认没有对进程邮箱大小设限制,所以当有大量消息持续发往某个进程时,会导致该进程邮箱过大,最终内存溢出并崩溃。
    在RabbitMQ中,如果生产者持续高速发送,而消费者消费速度较低时,如果没有流控,很快就会使内部进程邮箱大小达到内存阈值,阻塞生产者(得益于block机制,并不会崩溃)。然后RabbitMQ会进行page操作,将内存中的数据持久化到磁盘中。
    为了解决该问题,RabbitMQ使用了一种基于信用证的流控机制。消息处理进程有一个信用组{InitialCredit,MoreCreditAfter},默认值为{200, 50}。消息发送者进程A向接收者进程B发消息,每发一条消息,Credit数量减1,直到为0,A被block住;对于接收者B,每接收MoreCreditAfter条消息,会向A发送一条消息,给予A MoreCreditAfter个Credit,当A的Credit>0时,A可以继续向B发送消息。

    八、 RabbitMQ 多层消息队列

    RabbitMQ完全实现了AMQP协议,类似于一个邮箱服务。Exchange负责根据ExchangeType和RoutingKey将消息投递到对应的消息队列中,消息队列负责在消费者获取消息前暂存消息。在RabbitMQ中,MessageQueue主要由两部分组成,一个为AMQQueue,主要负责实现AMQP协议的逻辑功能。另外一个是用来存储消息的BackingQueue。
    为了高效处理入队和出队的消息、避免不必要的磁盘IO,BackingQueue进程为消息设计了4种状态和5个内部队列。
    (1) 4种状态包括:

    alpha,消息的内容和索引都在内存中;
    beta,消息的内容在磁盘,索引在内存;
    gamma,消息的内容在磁盘,索引在磁盘和内存中都有;
    delta,消息的内容和索引都在磁盘。
    
    • 1
    • 2
    • 3
    • 4

    对于持久化消息,RabbitMQ先将消息的内容和索引保存在磁盘中,然后才处于上面的某种状态(即只可能处于alpha、gamma、delta三种状态之一)。
    (2) 5个内部队列

    包括:q1、q2、delta、q3、q4。q1和q4队列中只有alpha状态的消息;q2和q3包含beta和gamma状态的消息;delta队列是消息按序存盘后的一种逻辑队列,只有delta状态的消息。所以delta队列并不在内存中,其他4个队列则是由erlang queue模块实现。
    这里写图片描述
    消息从q1入队,q4出队,在内部队列中传递的过程一般是经q1顺序到q4。实际执行并非必然如此:开始时所有队列都为空,消息直接进入q4(没有消息堆积时);内存紧张时将q4队尾部分消息转入q3,进而再由q3转入delta,此时新来的消息将存入q1(有消息堆积时)。

    当内存紧张时触发paging,paging将大量alpha状态的消息转换为beta和gamma;如果内存依然紧张,继续将beta和gamma状态转换为delta状态。Paging是一个持续过程,涉及到大量消息的多种状态转换,所以Paging的开销较大,严重影响系统性能。

    九、高可用队列(HA)
    在生产环境下,一般都不会允许rabbitmq这种消息中间件单点,以免单点故障导致服务不可用,那么rabbitmq同样可以集群部署来保证服务的可用性,在rabbitmq集群中,我们可以定义HA队列,可以在web管理平台设置,也可以通过AMQP接口设置,当我们定义某个HA队列的时候,会在集群的各个节点上都建立该队列,发布消息的时候,直接发送至master服务,当master服务受到消息后,把消息同步至各个从节点,假如开启事务的情况下,是需要在消息被同步到各个节点之后才算完成事务,所以会带来一定的性能损耗,所以还是回到之前说的,性能和安全直接,需要根据实际业务的需要找到平衡点。
    这里写图片描述
    当master服务宕机之后,其中一个slaver节点会升级为master,消息不会丢失(因为已经完成了事务的消息都会在各个节点有备份)
    ha-队列可以跨越集群的每台服务,或者仅使用其中一批独立节点。如果是全部节点都为副本的时候,将x-ha-policy参数设置为all,否则设置为nodes,然后在设置另一个参数:x-ha-nodes,该参数指定ha队列所在的节点列表。思考下,rabbitmq的集群节点是不是越多越好?

    展开全文
  • 本文实例讲述了PHP高级编程之消息队列原理与实现方法。分享给大家供大家参考,具体如下: 1. 什么是消息队列 消息队列(英语:Message queue)是一种进程间通信或同一进程的不同线程间的通信方式 2. 为什么使用消息...
  • 一、什么是消息队列 消息队列是在消息的传输过程中保存消息的容器,队列的主要目的是提供路由并保证消息的传递。 如果发送消息时接受者不可用,消息队列会保留消息,知道可以成功传递消息。 消息队列是一种异步的...

    原文章:https://www.cnblogs.com/middleware/p/9178734.html

    一、什么是消息队列

    消息队列是在消息的传输过程中保存消息的容器,队列的主要目的是提供路由并保证消息的传递。 如果发送消息时接受者不可用,消息队列会保留消息,知道可以成功传递消息。

    消息队列是一种异步的服务间通信方式,是分布式系统中重要的组件,主要解决应用耦合,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构。使用较多的消息队列:ActiveMQ,RocketMQ,RabbitMQ,Kafka等。

    二、分布式消息队列的原理

    在消息队列的基础上,稳定可靠的消息队列服务----分布式消息服务应运而生。
    分布式消息服务(DMS): 一项基于高可用分布式集群技术的消息中间件服务,提供普通队列、有序队列、kafka队列、ActiveMQ、RabbitMQ,兼容HTTP、TCP、AMQP协议,为分布式应用提供低延迟、高并发的异步通信机制。

    消息传递过程如下:

    ①生产者发送消息给消息队列:
    在这里插入图片描述
    生产者将消息M发送到队列中,消息M在队列中冗余分布,存在多个副本。

    ②消费者从消息队列中消费消息:

    加粗样式
    消费消息的一方叫消息消费者,通过调用消息服务的消费接口从队列中读取消息。 消费者从队列中得到消息M,在消费者消费消息的期间,消息M仍存在于消费队列中,但是消息M从消费者开始消费M开始的30秒内不能被该消费组再次进行消费。若在30秒内,该消息没有被消费者确认为消费成功,则DMS认为消息M未消费成功,将可以继续进行消费。

    ③消费者确认消息已经消费:

    在这里插入图片描述

    消费者确认消息消费完成,消息M该不再被该消费者所在消费组消费。消息M仍然保持在队列中,并且可以被其他消费组进行消费,消息在队列中至少存在72小时(除非队列删除),72小时后将会被删除。

    展开全文
  • 消息队列原理和应用场景总结同步架构和异步架构1.同步架构和异步架构的区别1.1 同步架构2.异步架构的主要组成部分:消息生产者、消息消费者、分布式消息队列3.异步架构的两种主要模型:点对点模型和发布订阅模型新的...
  • SpringCloud之RabbitMQ消息队列原理及配置  本篇章讲解RabbitMQ的用途、原理以及配置,RabbitMQ的安装请查看SpringCloud之RabbitMQ安装 一、MQ用途  1、同步变异步消息  场景:用户下单完成后,发送邮件和...
  • 消息队列原理总结

    千次阅读 2018-02-09 17:40:30
    消息队列应用的场景 业务解耦:消息队列要解决的最本质问题,实现设计的单一性原则,不耦合其他模块的业务。 最终一致性:用来处理延迟不那么敏感的“分布式事务”场景或者不重要的业务。 广播:下游有很多系统...
  • 最近,由于写作课程《分布式中间件实践之路》的契机,我再次阅读了这篇文章,觉得存在很多不足,对不起 “深入浅出” 一词,鉴于此,我决定再写一篇长文,详细解读基于 Kafka 和 ZooKeeper 的分布式消息队列原理,敬...
  • 消息队列原理笔记

    2019-04-25 12:56:00
    1.消息生产者、消费消费者、队列 消息生产者producer 发送消息队列 消费消费者consumer 从队列接受消息 broker MQ服务端,把消息从发送端传送到接收端 队列queue 消息存储区域。消息按照顺序发送接收,一旦消息被...
  • 深入理解Android消息队列原理

    千次阅读 2014-08-25 10:03:59
    本篇深入理解Android消息队列原理篇,将为您讲述Android世界中的消息机制,无论是android线程内部或者线程之间,当它们进行信息交互时,则会传递这些消息,为此,深入理解这些消息队列原理,将有助于我们更好的掌握...
  • C#消息队列原理

    千次阅读 2018-05-04 10:48:59
    使用消息队列的应用比较广的就是分布式开发,加入你有n个客户端,每个客户端都向服务器发送了一条对数据库某字段的操作,但不需要对数据库进行修改,这时候就可以用到消息队列。只要保证消息格式布标,消息的发送方...
  • 消息队列原理及选型

    2019-12-16 18:07:48
    什么是消息队列 消息队列(Message Queue)是一种进程间通信或同一进程的不同线程间的通信方式。 什么时候需要消息队列 异步处理:例如短信通知、终端状态推送、App推送、用户注册等 有些业务不想也不需要立即...
  • 1、RabbitMQ消息队列原理剖析 RabbitMQ 是具有代表性的开源消息中间件,当前较多地应用于企业系统内,用于对数据一致性、稳定性和可靠性要求较高的场景中。 RabbitMQ使用Erlang语言来编写的, 并且RabbitMQ是基于AMQP...
  • rabbitMQ消息队列原理

    2017-10-30 17:06:00
    MQ:Message Queue,消息队列,是一种应用程序对应用程序的通信方法;应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来链接它们。 1rabbitMQ入门及原理 rabbitMQ官网:...
  • RabbitMQ消息队列原理

    2019-08-13 11:23:07
    RabbitMQ教程 --------------------- 版权声明:本文为CSDN博主「niaobirdfly」的原创文章,遵循CC 4.0 by-sa版权协议,转载请附上原文出处链接及本声明。...消息队列(Message Queue,简称MQ)...
  • Android消息队列原理

    2016-04-02 16:04:41
    安卓的消息队列
  • 运用消息队列之后,用户下单完之后,下单信息写入数据库,再写入消息队列,发送邮件和发送短信各自去消息队列进行读取,节省时间,提高效率。 2、应用解耦 场景:用户下单后,订单系统需要多渠道通知用户。 a、下单...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 4,959
精华内容 1,983
关键字:

消息队列原理