RabbitMQ는 erlang에서 개발한 AMQP(Advanced Message Queuing Protocol)의 오픈 소스 구현입니다.
AMQP: Advanced Message Queuing Protocol은 메시지 지향 미들웨어용으로 설계된 애플리케이션 계층 프로토콜의 개방형 표준입니다. 메시지 미들웨어는 주로 구성 요소 간의 분리에 사용되며 메시지 발신자는 메시지 소비자의 존재를 알 필요가 없으며 그 반대도 마찬가지입니다. AMQP의 주요 기능은 메시지 지향, 대기열 지향, 라우팅(지점 간 및 게시/구독 포함), 안정성 및 보안입니다. RabbitMQ는 오픈 소스 AMQP 구현입니다. 서버 측은 Erlang 언어로 작성되었으며 Python, Ruby, .NET, Java, JMS, C, PHP, ActionScript, XMPP, STOMP 등과 같은 다양한 클라이언트를 지원합니다. AJAX를 지원합니다. 분산 시스템에서 메시지를 저장하고 전달하는 데 사용되며 사용 용이성, 확장성 및 고가용성 측면에서 우수한 성능을 발휘합니다.
RabbitMQ는 원래 금융 시스템에서 유래되었으며 분산 시스템에서 메시지를 저장하고 전달하는 데 사용됩니다. 가용성, 확장성, 고가용성 측면에서 좋습니다. 구체적인 기능은 다음과 같습니다:
- 可靠性（Reliability） RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认。 - 灵活的路由（Flexible Routing） 在消息进入队列之前，通过 Exchange 来路由消息的。对于典型的路由功能，RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能，可以将多个 Exchange 绑定在一起，也通过插件机制实现自己的 Exchange 。 - 消息集群（Clustering） 多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker 。 - 高可用（Highly Available Queues） 队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。 - 多种协议（Multi-protocol） RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等等。 - 多语言客户端（Many Clients） RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等。 - 管理界面（Management UI） RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。 - 跟踪机制（Tracing） 如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。 - 插件机制（Plugin System） RabbitMQ 提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。
- Message 消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。 - Publisher 消息的生产者，也是一个向交换器发布消息的客户端应用程序。 - Exchange 交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。 - Routing Key 路由关键字,exchange根据这个关键字进行消息投递。 - Binding 绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。 - Queue 消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。 - Connection 网络连接，比如一个TCP连接。 - Channel 信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内地虚拟连接，AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。 - Consumer 消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。 - Virtual Host 虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。 - Broker 表示消息队列服务器实体。它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,
Exchange는 메시지를 배포할 때 유형에 따라 배포 전략이 다릅니다. 현재 다이렉트, 팬아웃, 주제, 헤더의 네 가지 유형이 있습니다. headers는 라우팅 키 대신 AMQP 메시지의 헤더와 일치합니다. 또한 헤더 스위치는 직접 스위치와 완전히 동일하지만 성능이 훨씬 더 나쁩니다. 이제 다른 세 가지 유형을 살펴보십시오. :
메시지에서 라우팅 키가 Binding의 바인딩 키와 일치하면 교환기는 해당 큐로 메시지를 보냅니다. 라우팅 키는 대기열 이름과 정확하게 일치합니다. 대기열이 스위치에 바인딩되어 있고 라우팅 키가 "dog"이어야 하는 경우 라우팅 키 "dog"로 표시된 메시지만 전달되지 않습니다. 전달되지 않으며 "dog.puppy" 또는 "dog.puppy"도 전달되지 않습니다. 정확히 일치하는 유니캐스트 모드입니다.
팬아웃 유형 교환기로 전송된 모든 메시지는 모든 바인딩된 대기열에 배포됩니다. 팬아웃 교환은 라우팅 키를 처리하지 않고 단순히 대기열을 교환에 바인딩합니다. 교환으로 전송된 각 메시지는 교환에 바인딩된 모든 대기열로 전달됩니다. 서브넷 브로드캐스트와 마찬가지로 서브넷의 각 호스트는 메시지 복사본을 받습니다. 팬아웃 유형은 메시지 전달 속도가 가장 빠릅니다.
topic 교환기는 패턴 매칭을 통해 메시지의 라우팅 키 속성을 할당하고 라우팅 키를 특정 패턴과 일치시킵니다. 이때 대기열은 패턴에 바인딩되어야 합니다. 라우팅 키와 바인딩 키의 문자열을 점으로 구분된 단어로 분할합니다. 또한 두 개의 와일드카드 문자인 "#" 기호와 "" 기호도 인식합니다. #0개 이상의 단어를 일치시키세요. 단 하나의 단어만 일치하세요.
ConnectionFactory, Connection, Channel은 RabbitMQ에서 제공하는 API의 가장 기본적인 객체입니다.
RabbbitMQ의 분배 메커니즘은 확장에 매우 적합하며, 이제 부하가 더 커지면 특별히 설계되었습니다. create more Consumer는 작업 처리를 수행합니다
실제 응용에서는 소비자가 Queue에서 메시지를 수신하지만 처리가 완료되기 전에 충돌(또는 기타 사고)이 발생할 수 있습니다. 메시지가 손실되었습니다. 이러한 상황을 방지하기 위해 소비자가 메시지를 사용한 후 RabbitMQ에 영수증을 보내도록 요구할 수 있습니다. RabbitMQ는 메시지 확인(메시지 확인)을 받은 후에만 대기열에서 메시지를 제거합니다. 소비자의 RabbitMQ 연결이 끊어진 것으로 감지되면 RabbitMQ는 처리를 위해 다른 소비자(여러 소비자가 있는 경우)에게 메시지를 보냅니다. 여기에는 시간 초과 개념이 없습니다. 소비자가 메시지를 처리하는 기간에 관계없이 RabbitMQ 연결이 끊어지지 않는 한 메시지가 다른 소비자에게 전송되지 않습니다. 여기에서 또 다른 문제가 발생합니다. 개발자가 비즈니스 로직을 처리한 후 RabbitMQ에 영수증을 보내는 것을 잊어버리면 심각한 버그가 발생합니다. 소비자가 다시 시작된 후 점점 더 많은 메시지가 대기열에 축적되고 이러한 메시지를 반복적으로 소비하고 반복적으로 실행됩니다. 비즈니스 로직...
게다가, pub 메시지에는 ack가 없습니다.
如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。
要持久化队列queue的持久化需要在声明时指定durable=True;
这里要注意,队列的名字一定要是Broker中不存在的,不然不能改变此队列的任何属性.
队列和交换机有一个创建时候指定的标志durable,durable的唯一含义就是具有这个标志的队列和交换机会在重启之后重新建立,它不表示说在队列中的消息会在重启后恢复
消息持久化包括3部分
- 1.exchange持久化,在声明时指定durable => true hannel.ExchangeDeclare(ExchangeName,"direct",durable:true,autoDelete:false,arguments:null);//声明消息队列，且为可持久化的 - 2.queue持久化,在声明时指定durable => true channel.QueueDeclare(QueueName,durable:true,exclusive:false,autoDelete:false,arguments:null);//声明消息队列，且为可持久化的 - 3.消息持久化,在投递时指定delivery_mode => 2(1是非持久化). channel.basic_publish(exchange='', routing_key="task_queue", body=message, properties=pika.BasicProperties( delivery_mode = 2, # make message persistent ))
如果exchange和queue都是持久化的,那么它们之间的binding也是持久化的,如果exchange和queue两者之间有一个持久化，一个非持久化,则不允许建立绑定.
注意：一旦创建了队列和交换机,就不能修改其标志了,例如,创建了一个non-durable的队列,然后想把它改变成durable的,唯一的办法就是删除这个队列然后重现创建。
关于持久化的进一步讨论： 为了数据不丢失，我们采用了： 在数据处理结束后发送ack，这样RabbitMQ Server会认为Message Deliver 成功。 持久化queue，可以防止RabbitMQ Server 重启或者crash引起的数据丢失。 持久化Message，理由同上。 但是这样能保证数据100%不丢失吗？答案是否定的。问题就在与RabbitMQ需要时间去把这些信息存到磁盘上，这个time window虽然短，但是它的确还是有。在这个时间窗口内如果数据没有保存，数据还会丢失。还有另一个原因就是RabbitMQ并不是为每个Message都做fsync：它可能仅仅是把它保存到Cache里，还没来得及保存到物理磁盘上。因此这个持久化还是有问题。但是对于大多数应用来说，这已经足够了。当然为了保持一致性，你可以把每次的publish放到一个transaction中。这个transaction的实现需要user defined codes。那么商业系统会做什么呢？一种可能的方案是在系统panic时或者异常重启时或者断电时，应该给各个应用留出时间去flash cache，保证每个应用都能exit gracefully。
你可能也注意到了,分发机制不是那么优雅,默认状态下,RabbitMQ将第n个Message分发给第n个Consumer。n是取余后的,它不管Consumer是否还有unacked Message，只是按照这个默认的机制进行分发.
那么如果有个Consumer工作比较重,那么就会导致有的Consumer基本没事可做,有的Consumer却毫无休息的机会,那么,Rabbit是如何处理这种问题呢?
前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。
通过basic.qos方法设置prefetch_count=1，这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message，换句话说,在接收到该Consumer的ack前,它不会将新的Message分发给它
1
channel.basic_qos(prefetch_count=1)
注意，这种方法可能会导致queue满。当然，这种情况下你可能需要添加更多的Consumer，或者创建更多的virtualHost来细化你的设计。
RabbitMQ使用ProtoBuf序列化消息,它可作为RabbitMQ的Message的数据格式进行传输,由于是结构化的数据,这样就极大的方便了Consumer的数据高效处理,当然也可以使用XML，与XML相比, ProtoBuf有以下优势:
1.简单
2.size小了3-10倍
3.速度快了20-100倍
4.易于编程
6.减少了语义的歧义.
，ProtoBuf具有速度和空间的优势，使得它现在应用非常广泛
MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。 但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。
RabbitMQ 中实现RPC 的机制是：
客户端发送请求（消息）时，在消息的属性（MessageProperties ，在AMQP 协议中定义了14中properties ，这些属性会随着消息一起发送）中设置两个值replyTo （一个Queue 名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue 中）和correlationId （此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
服务器端收到消息并处理
服务器端处理完消息后，将生成一条应答消息到replyTo 指定的Queue ，同时带上correlationId 属性
客户端之前已订阅replyTo 指定的Queue ，从中收到服务器的应答消息后，根据其中的correlationId 属性分析哪条请求被执行了，根据执行结果进行后续业务处理
현재 인터넷 정보에 따르면 RabbitMQ, activeM, ZeroMQ 중에서 전체적으로 RabbitMQ가 첫 번째 선택입니다.
ZeroMq는 지원하지 않지만 ActiveMq와 RabbitMq는 모두 지원합니다. 지속성 메시지는 주로 불가항력 요인으로 인해 당사 시스템이 중단된 경우에도 메시지가 손실되지 않는 메커니즘을 나타냅니다.
신뢰성, 유연한 라우팅, 클러스터링, 트랜잭션, 고가용성 대기열, 메시지 정렬, 문제 추적, 시각적 관리 도구, 플러그인 시스템 등을 달성하기 위한 포괄적인 기술
RabbitMq/Kafka가 최고이고 ActiveMq가 두 번째이며 ZeroMq가 최악입니다. 물론 ZeroMq도 가능하지만 이를 구현하려면 수동으로 코드를 작성해야 하고 코드의 양도 적지 않다. 특히 신뢰성 측면에서는 내구성, 배송 확인, 게시자 확인, 고가용성 등이 있습니다.
RabbitMQ는 구현 언어가 본질적으로 높은 동시성 및 고가용성을 갖춘 Erlang 언어이기 때문에 가장 높습니다.
RabbitMq는 가용성, 안정성 및 신뢰성 측면에서 (이론적으로) RabbitMq가 더 좋습니다.
또한 Kafka의 포지셔닝은 주로 로그 등에 있습니다. Kafka의 원래 설계 의도는 로그를 처리하는 것이므로 로그(메시지) 시스템의 중요한 구성 요소로 간주할 수 있으므로 타겟이 높습니다. 비즈니스 측면이라면 RabbitMq를 선택하세요.
또한 Kafka의 성능(처리량, TPS)은 RabbitMq보다 훨씬 높습니다.
최종 선택 요약:
우리 시스템에 Kafka 또는 RabbitMq 옵션이 이미 있고 현재 비즈니스를 완벽하게 충족할 수 있다면 휠을 반복적으로 추가하고 재발명할 필요가 없는 것이 좋습니다.
Kafka와 RabbitMq 중에서 팀과 비즈니스에 맞는 것을 선택할 수 있다는 점이 가장 중요합니다. 그러나 모든 요소를 고려했을 때 현 단계에서는 세 번째 옵션이 없다는 점에는 의심의 여지가 없습니다.
더 많은 Laravel 관련 기술 기사를 보려면 Laravel Tutorial 칼럼을 방문하여 알아보세요!
위 내용은 RabbitMQ의 애플리케이션 시나리오 및 기본 원리의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
