Rabbit MQ基础概念

柔光的暖阳◎ 2022-05-24 13:38 214阅读 0赞

生产者(producer)创建消息，然后发布到代理服务器(RabbitMQ)。消息包含两部分内容：有效载荷(payload)和标签(label)。有效载荷就是你想要传输的数据，它可以是任何内容，标签描述了有效载荷，并且RabbitMQ用它来决定谁将获得消息的拷贝。

在应用程序和Rabbit代理服务器之间创建一条TCP连接，一旦TCP连接打开，应用程序就可以创建一条AMQP信道。信道是建立在“真实的”TCP连接内的虚拟连接。AMQP命令都是通过信道发送出去的。每条信道都会被指派一个唯一的ID。不论是发送消息，订阅队列货接收消息，这些动作都是通过信道完成的。

AMQP消息路由有三部分组成：交换器，队列和绑定。生产者把消息发布到交换器上；消息最终到达队列，并被消费者接收；绑定决定了消息如何从路由器到特定的队列。消费者通过以下两种方式从特定的队列中接收消息：

通过AMQP的baisc.consume命令订阅。这样会将信道置为接收模式，直到取消对队列的订阅为止。

向队列请求单条消息是通过AMQP的basic.get命令实现的。这样可以让消费者接收队列中的下一条消息。basic.get会订阅消息，获得单条消息，然后取消订阅。消费者应该使用basic.consume来实现高吞吐量。

当Rabbit队列拥有多个消费者时，队列收到的消息将以循环(round-robin)的方式发送给消费者。每条消息只会发送一个订阅的消费者。消息者通过AMQP的basic.ack命令显式地向RabbitMQ发送一个确认，或在订阅到队列的时候将auto\_ack参数设置为true。当设值了auto\_ack时，一旦消息者接收消息时，RabbitMQ会自动视为其确认了消息。

若处理消息的时候遇到了不可恢复的错误，但由于硬件问题，只影响到当前的消费者。只要消息尚未确认，则有两个选择：

把消费者从RabbitMQ服务器断开连接，此时RabbitMQ自动重新把消息入队并发送给另一消费者。缺点时连接/断开的方式会增加RabbitMQ的负担

使用AMQP的basic.reject命令。若把reject命令的requeue参数设置为true，RabbitMQ会将消息重新发送给下一个订阅的消费者。

消费者和生产者都可以使用AMQP的queue.declare命令来创建队列。若消费者在同一条信道上订阅了另一个队列的话，则无法在声明队列了，必须先取消订阅，将信道置为“传输”模式。当创建队列时，需要制定队列名称。消费者订阅队列时需要队列名称，并在绑定时也需要指定队列名称。若不指定队列名称，Rabbit会分配一个随机名称并在queue.declare命令的响应中返回。在创建队列时设置exclusive为true时，队列将变为私有的。当最后一个消费者取消订阅时，使用auto-delete参数可以让队列自动移除。当想检测队列是否存在时，可以设置queue.declare的passive选项为true。在该设置下，若该队列存在，queue.declare命令会返回成功，否则命令不会创建一个队列而返回一个错误。

队列是AMQP消息通信的基础模块：为消息提供了住处，消息在此等待消费；对负载均衡来说，只需附加消费者，并让RabbitMQ以循环的方式均匀地分配发来的消息；队列是Rabbit中消息的最后的终点。

当你想要将消息投递到队列时，通过把消息发送给交换器来完成。然后根据确定的规则，RabbitMQ将会决定消息该投递到哪个队列。这些规则被称为路由键(routing key)。队列通过路由键绑定到交换器。当你把消息发送到代理服务器时，消息将拥有一个路由键，即使路由键时空的，RabbitMQ也会将其和绑定使用的路由键进行匹配。若匹配，则该消息将投递到队列，若不匹配，消息将进入“黑洞”。

协议中定义了四种不同类型交换器：direct，fanout，topic和headers。

headers交换器：允许你匹配AMQP消息的header而非路由键。

direct交换器：如果路由键匹配，消息就被投递到对应的队列。服务器必须实现direct类型交换器，包含一个空白字符串名称的默认交换器，当声明一个队列时，它会自动绑定到默认交换器，并以队列名称作为路由键。

channel−>basicpublish(  c h a n n e l − > b a s i c p u b l i s h ( msg, ”, ‘queue-name’)；//第一个参数是发送信息；第二个参数exchange，此处是空的字符串，指定了默认交换器；第三个参事是路由键。

fanout交换器：将收到的消息广播到绑定的队列上。当发送一条消息到fanout交换器时，它会把消息投递到所有附加在此交换器上的队列。

topic交换器：使来自不同源头的消息能够到达同一个队列。

channel−>basicpublish(  c h a n n e l − > b a s i c p u b l i s h ( msg, ‘logs-exchange’, ‘error.msg-index’);

$channel->basic\_bind(‘msg-inbox-errors’, ‘logs-exchange’, ‘error.msg-index’);

$channel->basic\_bind(‘msg-index-logs’, ‘logs-exchange’, ‘\*.msg-inbox’);

每一个RabbitMQ服务器都能创建虚拟消息服务器，我们称之为虚拟主机(vhost)。每一个vhost本质上是一个mini版的RabbitMQ服务器，拥有自己的队列，交换器和绑定。它有自己的权限机制，这使得你能够安全地使用一个RabbitMQ服务器来服务众多应用程序。vhost通过在各个实例间提供逻辑上的分离，允许为不同应用程序安全保密地运行数据。当在RabbitMQ创建一个用户时，用户通常会被指派给至少一个vhost，并且只能访问vhost内的队列，交换器和绑定。当在RabbitMQ集群上创建vhost时，整个集群上都会创建该vhost。vhost不仅消除了为基础架构中的每一层运行一个RabbitMQ服务器的需要，同样也避免了为每一层创建不同集群。通过使用rabbitmqctl add\_vhost vhost\_name来创建一个vhost。通过rabbitmqctl delete\_vhost vhost-name来删除vhost。通过rabbitmqctl list\_vhosts来显式所有的vhost。

队列和交换器的durable属性默认为false。当RabbitMQ崩溃或重启后所有数据将会丢失。因此应该将队列和交换器设置为true。能从AMQP服务器崩溃中恢复的消息称之为持久化消息。在消息发布前，通过把它的“投递模式(delivery mode)”设置为2来把消息标记成持久化。因此，若消息要从Rabbit崩溃中恢复，需要把它的投递模式选项设置为2；发送到持久化的交换器；到达持久化的队列。RabbitMQ确保持久性消息能从服务器重启中恢复的方式是，将它们写入磁盘上的一个持久化日志文件。当发布一条持久性信息导持久交换器上时，Rabbit会在消息提交到日志文件后才发送响应。使用持久化机制会导致性能的下降，因为要不断将消息写入磁盘同时会极大减少RabbitMQ服务器每秒可处理的消息总数。持久性消息在RabbitMQ内建集群环境下工作得并不好，若集群中某个节点崩溃了，则这个队列也就从整个集群中消失了。若节点宕机，则其他队列也无法使用了。若要求单台Rabbit服务器每秒处理10w消息，生产者可以在单独的信道上监听应答队列。若消息应答未在合理时间范围内到达，生产者可以重新发送消息。即要保证消息的投递决定了相对于其他类型的消息会有更低的吞吐量。

在AMQP中，在把信道设置成事务模式后，通过信道发送那些想要确认的消息之后还有很多其他AMQP命令，这些命令的执行与否取决于第一条消息发送是否成功。一旦发送完所有命令，就可以提交事务了。若事务中的首次发布成功了，那么信道会在事务中完成其他AMQP命令，若失败，则其他AMQP命令将不会执行。使用事务会造成严重的性能下降，降低2～10倍的消息吞吐量并且使生产者应用程序产生同步。因此，RabbitMQ提出了发送方确认模式，即将信道设置为confirm模式。一旦信道进入confirm模式，所有在信道上发布的消息会被指派唯一一个ID号(由于一条信道只能被耽单个程使用，因而可以确保信道上发布的消息是连续的。因此RabbitMQ做了个简单的假设：任一信道上发布的消息将获得ID1，并且信道接下来的每一条消息的ID都将步进1。每次应用程序信道发布消息时，需要把计数器加1)。一旦消息被投递给所有匹配的队列后，信道会发送一个发送方确认模式给生产者应用程序使得生产者知晓消息已经到达队列了。发送方确认的好处是移步，生产者可以在等待确认的同时继续发送下一条。当确认消息最终受到时，生产者应用的回调方法会被触发来处理确认信息。若Rabbit发生了内部错误而导致信息丢失，Rabbit会发送nack消息。由于没有回滚，发送方确认模式更加轻量级，同时对Rabbit代理服务器的性能影响几乎可以忽略不计。