Rabbit MQ

相关文章

• 我为什么要选择RabbitMQ ，RabbitMQ简介，各种MQ选型对比 https://www.sojson.com/blog/48.html
• 消息队列--RabbitMQ基础 https://blog.csdn.net/u012482647/article/details/88537627
• RabbitMQ流控实测 https://blog.csdn.net/qq_20892953/article/details/80593282

RMQ和相关插件默认端口

• 4369 (epmd), 25672 (Erlang distribution)
• 5672, 5671 (AMQP 0-9-1 without and with TLS) (高级消息队列协议，rmq默认使用的协议)
• 15672 (if management plugin is enabled) ( web 管理界面端口 通过 http://localhost:15672访问)
• 61613, 61614 (if STOMP is enabled) (通讯协议)
• 1883, 8883 (if MQTT is enabled) (通讯协议)

版本号	端口号
epmd	4369
Erlang distribution	25672
AMQP 0-9-1 without and with TLS	5672, 5671
if management plugin is enabled	15672
if STOMP is enabled	61613, 61614
if MQTT is enabled	1883, 8883

插件 rabbitmq-management

rabbitmq-management是RabbitMq web管理端，用的是erlang的cowboy框架进行开发，默认端口是15672 。

empd

[Erlang 0123] Erlang EPMD https://www.cnblogs.com/me-sa/p/erlang-epmd.html

epmd进程和Erlang节点进程如影随形,在Rabbitmq集群,Ejabberd集群,Couchbase集群产品文档中都会有相当多的内容讲epmd,epmd是什么呢?

epmd 是Erlang Port Mapper Daemon的缩写,全称足够明确表达它的功能了(相比之下,OTP就是一个难以从字面理解的名字);epmd完成Erlang节点和IP,端口的映射关系

AMQP 0-9-1without and with TLS

AMQP-0-9-1中文规范 http://www.blogjava.net/qbna350816/archive/2016/08/12/431554.html

AMQP 0-9-1模型简介（官网直译） https://www.cnblogs.com/xiaochengzi/p/6895126.html

AMQP 0-9-1（Advanced Message Queuing Protocol）高级消息队列协议是一个消息协议，它支持符合标准的客户端请求程序与符合标准的消息中间件代理进行通信。

STOMP

STOMP is the Simple (or Streaming) Text Orientated Messaging Protocol.

http://stomp.github.io

MQTT

MQTT is a machine-to-machine (M2M)/"Internet of Things" connectivity protocol.

http://mqtt.org

AMQP

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。 AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。 RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

QPID

http://qpid.2158936.n2.nabble.com

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。

Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。

ConnectionFactory为Connection的制造工厂。

Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

Connection

Connection的三种状态

• running:运行中
• idle:空闲
• flow:大量数据未消费的时候的状态

参考链接：

RabbitMq系列之三：web管理端 https://www.jianshu.com/p/7b6e575fd451

[rabbitmq-discuss] Connections in state "flow" http://lists.rabbitmq.com/pipermail/rabbitmq-discuss/2012-May/020302.html

https://grokbase.com/t/rabbitmq/rabbitmq-discuss/1478x8txyr/connection-is-always-in-flow-state

Broker

Broker（消息代理）: 接收和分发消息的应用，RabbitMQ Server就是Message Broker，实现消息队列和路由功能的过程

Consumer

Consumer（消息消费者）：一个从消息队列中取得消息的客户端用用程序

Queue

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。 这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

另外pub message是没有ack的。

Message durability

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

Prefetch count

前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

Exchange

在上一节我们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

Exchange是按照什么逻辑将消息路由到Queue的？这个将在Binding一节介绍。 RabbitMQ中的Exchange有四种类型，不同的类型有着不同的路由策略，这将在Exchange Types一节介绍。

routing key

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。 在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。 RabbitMQ为routing key设定的长度限制为255 bytes。

Binding

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

Binding key

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。 在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。 binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

Exchange Types

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

fanout

fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

direct

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。

以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；

如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

topic

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

• routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如 “stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
• binding key与routing key一样也是句点号“. ”分隔的字符串
• binding key中可以存在两种特殊字符“*”与“#”，用于做模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）

以上图中的配置为例，

routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，

routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，

routingKey=”lazy.brown.fox”的消息会路由到Q2，

routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；

routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。

headers

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。 在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。 该类型的Exchange没有用到过（不过也应该很有用武之地），所以不做介绍。

RPC

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属性分析哪条请求被执行了，根据执行结果进行后续业务处理

VirtualHost

VirtualHost只是起到一个命名空间的作用，所以可以多个user共同使用一个virtual host，VirtualHost = "/"是系统默认的，就是说当我们创建一个到rabbitmq的connection时候，它的命名空间是'/'，需要注意的是不同的命名空间之间的资源是不能访问的，比如 exchang,queue,bingding等。

总结

本文介绍了RabbitMQ 中个人认为最重要的概念，充分利用RabbitMQ 提供的这些功能就可以处理我们绝大部分的异步业务了。

RabbitMQ 选型和对比

1.从社区活跃度

按照目前网络上的资料，RabbitMQ 、activeM 、ZeroMQ 三者中，综合来看，RabbitMQ 是首选。

2.持久化消息比较

ZeroMq 不支持，ActiveMq 和RabbitMq 都支持。持久化消息主要是指我们机器在不可抗力因素等情况下挂掉了，消息不会丢失的机制。

3.综合技术实现

可靠性、灵活的路由、集群、事务、高可用的队列、消息排序、问题追踪、可视化管理工具、插件系统等等。

RabbitMq / Kafka 最好，ActiveMq 次之，ZeroMq 最差。当然ZeroMq 也可以做到，不过自己必须手动写代码实现，代码量不小。尤其是可靠性中的：持久性、投递确认、发布者证实和高可用性。

4.高并发

毋庸置疑，RabbitMQ 最高，原因是它的实现语言是天生具备高并发高可用的erlang 语言。

5.比较关注的比较， RabbitMQ 和 Kafka

RabbitMq 比Kafka 成熟，在可用性上，稳定性上，可靠性上， RabbitMq 胜于 Kafka （理论上）。

另外，Kafka 的定位主要在日志等方面， 因为Kafka 设计的初衷就是处理日志的，可以看做是一个日志（消息）系统一个重要组件，针对性很强，所以 如果业务方面还是建议选择 RabbitMq 。

还有就是，Kafka 的性能（吞吐量、TPS ）比RabbitMq 要高出来很多。

选型最后总结：

如果我们系统中已经有选择 Kafka ，或者 RabbitMq ，并且完全可以满足现在的业务，建议就不用重复去增加和造轮子。

可以在 Kafka 和 RabbitMq 中选择一个适合自己团队和业务的，这个才是最重要的。但是毋庸置疑现阶段，综合考虑没有第三选择。

RabbitMQ管理

Rabbitmq服务器的主要通过rabbitmqctl和rabbimq-plugins两个工具来管理，以下是一些常用功能。

1）. 服务器启动与关闭

​ 启动: rabbitmq-server –detached

​ 关闭:rabbitmqctl stop

​ 若单机有多个实例，则在rabbitmqctlh后加–n 指定名称

2）. 插件管理

​ 开启某个插件：rabbitmq-pluginsenable xxx

​ 关闭某个插件：rabbitmq-pluginsdisablexxx

​ 注意：重启服务器后生效。

3）.virtual_host管理

​ 新建virtual_host: rabbitmqctladd_vhost xxx

​ 撤销virtual_host:rabbitmqctl delete_vhost xxx

4）. 用户管理

​ 新建用户：rabbitmqctl add_user xxxpwd

​ 删除用户: rabbitmqctl delete_user xxx

​ 改密码: rabbimqctlchange_password {username} {newpassword}

​ 设置用户角色：rabbitmqctlset_user_tags {username} {tag ...}

​ Tag可以为 administrator,monitoring, management

5）. 权限管理

​ 权限设置：set_permissions [-pvhostpath] {user} {conf} {write} {read}

​ Vhostpath

​ Vhost路径

​ user

​ 用户名

​ Conf

​ 一个正则表达式match哪些配置资源能够被该用户访问。

​ Write

​ 一个正则表达式match哪些配置资源能够被该用户读。

​ Read

​ 一个正则表达式match哪些配置资源能够被该用户访问。

6）. 获取服务器状态信息

​ 服务器状态：rabbitmqctl status

​ 队列信息：rabbitmqctl list_queues[-p vhostpath] [queueinfoitem ...]

​ Queueinfoitem可以为：name，durable，auto_delete，arguments，messages_ready，

​ messages_unacknowledged，messages，consumers，memory

​ Exchange信息：rabbitmqctllist_exchanges[-p vhostpath] [exchangeinfoitem ...]

​ Exchangeinfoitem有：name，type，durable，auto_delete，internal，arguments.

​ Binding信息：rabbitmqctllist_bindings[-p vhostpath] [bindinginfoitem ...]

​ Bindinginfoitem有：source_name，source_kind，destination_name，destination_kind，routing_key，arguments

​ Connection信息：rabbitmqctllist_connections [connectioninfoitem ...]

​ Connectioninfoitem有：recv_oct，recv_cnt，send_oct，send_cnt，send_pend等。

​ Channel信息：rabbitmqctl list_channels[channelinfoitem ...]

​ Channelinfoitem有consumer_count，messages_unacknowledged，messages_uncommitted，acks_uncommitted，messages_unconfirmed，prefetch_count，client_flow_blocked

部分内容参考自：http://blog.csdn.net/mlks_2008/article/details/18988301

rabbitmq配置

一般情况下，RabbitMQ的默认配置就足够了。如果希望特殊设置的话，有两个途径：

一个是环境变量的配置文件 rabbitmq-env.conf ；

一个是配置信息的配置文件 rabbitmq.config；

注意，这两个文件默认是没有的，如果需要必须自己创建。

rabbitmq-env.conf

这个文件的位置是确定和不能改变的，位于：/etc/rabbitmq目录下（这个目录需要自己创建）。

文件的内容包括了RabbitMQ的一些环境变量，常用的有：

#RABBITMQ_NODE_PORT= //端口号

#HOSTNAME=

RABBITMQ_NODENAME=mq

RABBITMQ_CONFIG_FILE= //配置文件的路径

RABBITMQ_MNESIA_BASE=/rabbitmq/data //需要使用的MNESIA数据库的路径

RABBITMQ_LOG_BASE=/rabbitmq/log //log的路径

RABBITMQ_PLUGINS_DIR=/rabbitmq/plugins //插件的路径

具体的列表见：http://www.rabbitmq.com/configure.html#define-environment-variables

rabbitmq.config

这是一个标准的erlang配置文件。它必须符合erlang配置文件的标准。

它既有默认的目录，也可以在rabbitmq-env.conf文件中配置。

文件的内容详见：http://www.rabbitmq.com/configure.html#config-items

消息队列--RabbitMQ基础

https://blog.csdn.net/u012482647/article/details/88537627

一、什么是消息队列？

消息队列（MessageQueue），简称MQ，FIFO，即先进先出，是一种为了解决“逻辑解耦+物理解耦”的消息通信服务。

MQ是一种系统间相互协作的通信机制， 在系统解耦、流量削峰、日志收集、保持事务一致性等方便都会有很大帮助，一个典型的消息队列，如下图所示：

• Broker：消息处理中心，负责消息的接受、存储、转发；
• Producer：消息生产者，负责产生和发送消息到消息处理中心；
• Consumer：消息消费者，负责从消息处理中心获取消息，并进行相应的处理；

二、什么是AMQP？

AMQP（Advanced Message Queue Protocol）是一种用于实现消息队列所涉及的协议，常见的协议有AMQP、MQTT、STOMP、XMPP等。RabbitMQ是一个由erlang开发的AMQP的开源实现。官网地址：RabbitMQ官网地址

先了解一下AMQP协议的基本概念：

1. Message（消息）：消息服务器处理数据的原子单元。消息包括一个内容头、一组属性和一个内容体。消息可以被存储到硬盘中，本身有优先级；
2. Publisher(消息生产者)：一个向Exchange(交换器)发布消息的客户端应用程序；
3. Broker（消息代理）: 接收和分发消息的应用，RabbitMQ Server就是Message Broker，实现消息队列和路由功能的过程
4. Consumer（消息消费者）：一个从消息队列中取得消息的客户端用用程序
5. Exchange（交换器）: Message到达Broker的第一站，根据分发规则，匹配查询表中的 Routing key，分发消息到Queue中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)，详解见第三部分。
6. Binding（绑定）: 用于消息队列和交换器之间的关联，binding中可以包含routing key。交换器就是一个由绑定构成的路由表
7. Virtual host（虚拟主机）: 出于多租户和安全因素设计的，把AMQP的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的vhost创建exchange／queue等，每一个虚拟主机本质上是一个mini版的消息服务器。
8. Routing key（路由规则）：虚拟机使用它来确定如何路由一个特定的消息
9. Connection（连接）: publisher／consumer和broker之间的TCP连接。断开连接的操作只会在client端进行，Broker不会断开连接，除非出现网络故障或broker服务出现问题。
10. Channel（信道）: 如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接，信道是一条独立的双向数据流通道，是建立在真实的TCP连接内的虚拟连接，一个connection可以包含多条信道；Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。
11. Queue(消息队列): 用来保存消息，等待consumer取走。一个message可以被同时拷贝到多个queue中。

AMQP的生命周期：

Publisher产生一条消息，发送到Broker，broker中的Exchange可以被理解为一个规则表，Broker收到消息以后根据Routing key查询投递目标queue，Consumer向Broker订阅消息时，会指定自己监听那个Queue，当有数据到达Queue时，Broker会推送数据到Consumer。

三、交换器类型

在AMQP生命周期图中，生产者产生的消息并不是直接发送给消息队列Queue的，而是要经过Exchange（交换器），由Exchange再将消息路由到一个或多个Queue，当然这里还会对不符合路由规则的消息进行丢弃掉，这里指的是后续要谈到的Exchange Type。

那么Exchange是怎样将消息准确的推送到对应的Queue的呢？那么这里的功劳最大的当Binding，RabbitMQ是通过Binding将Exchange和Queue链接在一起，这样Exchange就知道如何将消息准确的推送到Queue中去。

不同类型的交换器分发消息的策略是不同的，目前交换器有4种类型：Direct、Fanout、Topic、Headers，其中Headers交换器匹配AMQP消息的header而不是路由键，目前几乎不用了，所以我们来看一下另外三种交换器。

3.1-Direct

direct类型的Exchange路由规则很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中，路由键要和队列名称要完全匹配（并不是名字一样）Direct交换器是完全匹配、单播的模式。

3.2-Fanout

fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。这很像子网广播，子网内的每个主机都会获得一份复制的消息。

3.3-Topic

相对于Direct交换器，Topic交换器就是模糊查询。binding key中可以存在两种特殊字符“”与“#”，用于做模糊匹配，其中“”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）

在RabbitMQ默认定义一些交换机，主要如下：

3.4-默认交换机

RabbitMQ默认定义一些交换机

默认交换机（default exchange）实际上是一个由RabbitMQ预先声明好的名字为空字符串的直连交换机（direct exchange）。它有一个特殊的属性使得它对于简单应用特别有用处：那就是每个新建队列（queue）都会自动绑定到默认交换机上，绑定的路由键（routing key）名称与队列名称相同。

如：当你声明了一个名为”test”的队列，RabbitMQ会自动将其绑定到默认交换机上，绑定（binding）的路由键名称也是为”test”。因此，当消息（路由键为”test”）被发送到默认交换机时，此消息会被默认交换机路由至名为”test”的queue中。

类似amq.*的名称的交换机是RabbitMQ默认创建的交换机。这些队列名称被预留做RabbitMQ内部使用，不能被应用使用，否则抛出403 (ACCESS_REFUSED)错误。