一、MQ的基本概念
1 .1MQ概述
MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。
1.2MQ的优势和劣势
-
应用解耦
image.png -
异步提速
image.png -
削峰填谷
image.png
image.png
1.3常见MQ产品
1.4 RabbitMQ 简介
AMQP，即 Advanced Message Queuing Protocol（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。2006年，AMQP 规范发布。类比HTTP。
RabbitMQ 基础架构如下图：
⚫ Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker
⚫ Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网
络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多
个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
⚫ Connection：publisher／consumer 和 broker 之间的 TCP 连接
⚫ Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection极大减少了操作系统建立 TCP connection 的开销
⚫ Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)
⚫ Queue：消息最终被送到这里等待 consumer 取走
⚫ Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存
到 exchange 中的查询表中，用于 message 的分发依据
RabbitMQ 提供了 6 种工作模式：
简单模式、work queues、Publish/Subscribe 发布与订阅模式、Routing路由模式、Topics 主题模式、RPC 远程调用模式（远程调用，不太算 MQ；暂不作介绍）。
1.5 JMS
⚫ JMS 即 Java 消息服务（JavaMessage Service）应用程序接口，是一个 Java 平台中关于面向消息中间件的API
⚫ JMS 是 JavaEE 规范中的一种，类比JDBC
⚫ 很多消息中间件都实现了JMS规范，例如：ActiveMQ。RabbitMQ 官方没有提供 JMS 的实现包，但是开源社区有
二、RabbitMQ的安装和配置
- 安装依赖环境
- 安装Erlang
- 安装RabbitMQ
- 开启管理界面及配置
- 启动
- 配置虚拟主机及用户
三、RabbitMQ 的工作模式
3.1 Work queues工作队列模式
应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。
3.2 Pub/Sub 订阅模式
在订阅模型中，多了一个 Exchange 角色，而且过程略有变化：
Exchange有常见以下3种类型：
➢ Fanout：广播，将消息交给所有绑定到交换机的队列
➢ Direct：定向，把消息交给符合指定routing key 的队列
➢ Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！
3.3 Routing 路由模式
⚫ 队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）
⚫ 消息的发送方在向 Exchange 发送消息时，也必须指定消息的 RoutingKey
⚫ Exchange 不再把消息交给每一个绑定的队列，而是根据消息的 Routing Key 进行判断，只有队列的Routingkey 与消息的 Routing key 完全一致，才会接收到消息
3.4 Topics 通配符模式
⚫ Topic 类型与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列。只不过 Topic 类型Exchange 可以让队列在绑定 Routing key 的时候使用通配符！
⚫ Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert
⚫ 通配符规则：# 匹配一个或多个词，* 匹配不多不少恰好1个词，例如：item.# 能够匹配 item.insert.abc 或者 item.insert，item.* 只能匹配 item.insert
3.5 简单模式 HelloWorld
一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）。
四、Spring整合RabbitMQ
五、RabbitMQ高级特性
5.1 消息的可靠投递
RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。
⚫ confirm 确认模式
⚫ return 退回模式
rabbitmq 整个消息投递的路径为：
producer--->rabbitmq broker--->exchange--->queue--->consumer
⚫ 消息从 producer 到 exchange 则会返回一个 confirmCallback 。
⚫ 消息从 exchange-->queue 投递失败则会返回一个 returnCallback 。
我们将利用这两个 callback 控制消息的可靠性投递
5.2 Consumer Ack
ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。
有三种确认方式：
• 自动确认：acknowledge="none"
• 手动确认：acknowledge="manual"
• 根据异常情况确认：acknowledge="auto"，（这种方式使用麻烦，不作讲解）
其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。
5.3 消费端限流
➢ 在<rabbit:listener-container> 中配置 prefetch属性设置消费端一次拉取多少消息
➢ 消费端的确认模式一定为手动确认。acknowledge="manual"
5.4 TTL
➢ TTL 全称 Time To Live（存活时间/过期时间）。
➢ 当消息到达存活时间后，还没有被消费，会被自动清除。
➢ RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。
5.5 死信队列
死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX。
消息成为死信的三种情况：
- 队列消息长度到达限制；
- 消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；
- 原队列存在消息过期设置，消息到达超时时间未被消费；
队列绑定死信交换机：
给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key
5.6 延迟队列
延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。
很可惜，在RabbitMQ中并未提供延迟队列功能。
但是可以使用：TTL+死信队列 组合实现延迟队列的效果。
5.7 日志与监控
RabbitMQ日志
RabbitMQ默认日志存放路径： /var/log/rabbitmq/rabbit@xxx.log
日志包含了RabbitMQ的版本号、Erlang的版本号、RabbitMQ服务节点名称、cookie的hash值、RabbitMQ配置文件地址、内存限制、磁盘限制、默认账户guest的创建以及权限配置等等。web管控台监控
-
rabbitmqctl管理和监控
image.png
5.8 消息追踪
在使用任何消息中间件的过程中，难免会出现某条消息异常丢失的情况，这个时候就需要有一个较好的机制跟踪记录消息的投递过程，以此协助开发和运维人员进行问题的定位。
在RabbitMQ中可以使用Firehose和rabbitmq_tracing插件功能来实现消息追踪。
- forehose
firehose的机制是将生产者投递给rabbitmq的消息，rabbitmq投递给消费者的消息按照指定的格式发送到默认的exchange上。这个默认的exchange的名称为amq.rabbitmq.trace，它是一个topic类型的exchange。发送到这个exchange上的消息的routing key为 publish.exchangename 和deliver.queuename。其中exchangename和queuename为实际exchange和queue的名称，分别对应生产者投递到exchange的消息，和消费者从queue上获取的消息。
rabbitmqctl trace_on：开启Firehose命令
rabbitmqctl trace_off：关闭Firehose命令 - rabbitmq_tracing
rabbitmq_tracing和Firehose在实现上如出一辙，只不过rabbitmq_tracing的方式比Firehose多了一
层GUI的包装，更容易使用和管理。
启用插件：rabbitmq-plugins enable rabbitmq_tracing
六、应用问题
6.1 消息可靠性保障
6.2消息幂等性保障
幂等性指一次和多次请求某一个资源，对于资源本身应该具有同样的结果。也就是说，其任意多次执行对资源本身所产生的影响均与一次执行的影响相同。
在MQ中指，消费多条相同的消息，得到与消费该消息一次相同的结果。
七、RabbitMQ集群搭建
实际生产应用中都会采用消息队列的集群方案，如果选择RabbitMQ那么有必要了解下它的集群方案原理
7.1集群方案的原理
RabbitMQ这款消息队列中间件产品本身是基于Erlang编写，Erlang语言天生具备分布式特性（通过同步Erlang集群各节点的magic cookie来实现）。因此，RabbitMQ天然支持Clustering。这使得RabbitMQ本身不需要像ActiveMQ、Kafka那样通过ZooKeeper分别来实现HA方案和保存集群的元数据。集群是保证可靠性的一种方式，同时可以通过水平扩展以达到增加消息吞吐量能力的目的。
7.2 集群管理
rabbitmqctl join_cluster {cluster_node} [–ram]
将节点加入指定集群中。在这个命令执行前需要停止RabbitMQ应用并重置节点。
rabbitmqctl cluster_status
显示集群的状态。
rabbitmqctl change_cluster_node_type {disc|ram}
修改集群节点的类型。在这个命令执行前需要停止RabbitMQ应用。
rabbitmqctl forget_cluster_node [–offline]
将节点从集群中删除，允许离线执行。
rabbitmqctl update_cluster_nodes {clusternode}
在集群中的节点应用启动前咨询clusternode节点的最新信息，并更新相应的集群信息。这个和join_cluster不同，它不加入集群。考虑这样一种情况，节点A和节点B都在集群中，当节点A离线了，节点C又和节点B组成了一个集群，然后节点B又离开了集群，当A醒来的时候，它会尝试联系节点B，但是这样会失败，因为节点B已经不在集群中了。
rabbitmqctl cancel_sync_queue [-p vhost] {queue}
取消队列queue同步镜像的操作。
rabbitmqctl set_cluster_name {name}
设置集群名称。集群名称在客户端连接时会通报给客户端。Federation和Shovel插件也会有用到集群名称的地方。集群名称默认是集群中第一个节点的名称，通过这个命令可以重新设置。
3.4 RabbitMQ镜像集群配置
保证队列的高可用性
设置的镜像队列可以通过开启的网页的管理端Admin->Policies，也可以通过命令。
3.5 负载均衡-HAProxy
HAProxy提供高可用性、负载均衡以及基于TCP和HTTP应用的代理，支持虚拟主机，它是免费、快速并且可靠的一种解决方案,包括Twitter，Reddit，StackOverflow，GitHub在内的多家知名互联网公司在使用。HAProxy实现了一种事件驱动、单一进程模型，此模型支持非常大的并发连接数。
