20200923RabbitMQ的介绍使用分布式的内容整合总结

搜索与商品服务的问题

目前我们已经完成了商品详情和搜索系统的开发。思考一下&＃xff0c;是否存在问题&＃xff1f;

• 商品的原始数据保存在数据库中&＃xff0c;增删改查都在数据库中完成。
• 搜索服务数据来源是索引库&＃xff0c;如果数据库商品发生变化&＃xff0c;索引库数据不能及时更新。
• 商品详情做了页面静态化&＃xff0c;静态页面数据也不会随着数据库商品发生变化。

如果我们在后台修改了商品的价格&＃xff0c;搜索页面和商品详情页显示的依然是旧的价格&＃xff0c;这样显然不对。该如何解决&＃xff1f;

这里有两种解决方案&＃xff1a;

• 方案1&＃xff1a;每当后台对商品做增删改操作&＃xff0c;同时要修改索引库数据及静态页面
• 方案2&＃xff1a;搜索服务和商品页面服务对外提供操作接口&＃xff0c;后台在商品增删改后&＃xff0c;调用接口

以上两种方式都有同一个严重问题&＃xff1a;就是代码耦合&＃xff0c;后台服务中需要嵌入搜索和商品页面服务&＃xff0c;违背了微服务的独立原则。

所以&＃xff0c;我们会通过另外一种方式来解决这个问题&＃xff1a;消息队列

什么是消息队列

消息队列&＃xff0c;即MQ&＃xff0c;Message Queue&＃xff0c;消息队列是典型的&＃xff1a;生产者、消费者模型。生产者不断向消息队列中生产消息&＃xff0c;消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的&＃xff0c;而且只关心消息的发送和接收&＃xff0c;没有业务逻辑的侵入&＃xff0c;这样就实现了生产者和消费者的解耦结合前面所说的问题&＃xff0c;如果以后有其它系统也依赖商品服务的数据&＃xff0c;同样监听消息即可&＃xff0c;商品服务无需任何代码修改。

AMQP和JMS

MQ是消息通信的模型&＃xff0c;并不是具体实现。现在实现MQ的有两种主流方式&＃xff1a;AMQP、JMS

两者间的区别和联系&＃xff1a;

• JMS是定义了统一的接口&＃xff0c;来对消息操作进行统一&＃xff1b;AMQP是通过规定协议来统一数据交互的格式
• JMS限定了必须使用Java语言&＃xff1b;AMQP只是协议&＃xff0c;不规定实现方式&＃xff0c;因此是跨语言的。
• JMS规定了两种消息模型&＃xff1b;而AMQP的消息模型更加丰富

常见MQ产品

​ 消息队列中间件的比较

• RabbitMQ&＃xff1a;

  • 优点&＃xff1a;支持很多协议如&＃xff1a;AMQP&＃xff0c;XMPP&＃xff0c;STMP&＃xff0c;STOMP&＃xff1b;灵活的路由&＃xff1b;成熟稳定的集群方案&＃xff1b;负载均衡&＃xff1b;数据持久化等。
  • 缺点&＃xff1a;速度较慢&＃xff1b;比较重量级&＃xff0c;安装需要依赖Erlang环境。

• Redis&＃xff1a;

  • 分布式消息系统&＃xff0c;高吞吐量
  • 优点&＃xff1a;比较轻量级&＃xff0c;易上手
  • 缺点&＃xff1a;单点问题&＃xff0c;功能单一

• Kafka&＃xff1a;

  • 优点&＃xff1a;高吞吐&＃xff1b;分布式&＃xff1b;快速持久化&＃xff1b;负载均衡&＃xff1b;轻量级
  • 缺点&＃xff1a;极端情况下会丢消息

• ActiveMQ&＃xff1a;基于JMS

• RocketMQ&＃xff1a;基于JMS&＃xff0c;阿里巴巴产品&＃xff0c;目前交由Apache基金会

为什么选用了RabbitMQ,首先 基于AMQP协议&＃xff0c;稳定性好&＃xff0c;不规定实现方式&＃xff0c;可以跨语言&＃xff0c;可分布式部署&＃xff0c;并发量远高于ActiveMQ

准备阶段

下载

​ 因为是erlang编写的所以就像要安装jdk一样安装基础的erlang运行环境

​ 当然还有RabbitMQ安装文件

安装

配置环境变量

​ erlang和RabbitMQ需要配置

紧接着是安装可视化操作插件

下载插件的指令

C:\WINDOWS\system32>rabbitmq-plugins enable rabbitmq_management

进入RabbitMQ的管理软件页面

http://localhost:15672/

默认账号密码&＃xff1a;

guest

guest

这个时候注意一点一定不要使用默认的账号去操作RabbitMQ&＃xff0c;这样会导致无法连接的意外情况。

所以要自己新建一个账户&＃xff0c;因为是用来做测试&＃xff0c;为了方便权限就直接全部分配就好

RabbitMQ6种消息模型

RabbitMQ提供了6种消息模型&＃xff0c;但是第6种其实是RPC&＃xff0c;并不是MQ&＃xff0c;我们平时用不到所以就没有了解

点对点&＃xff1a;一个生产者生产的消息&＃xff0c;只能由一个消费者处理

普通消息模型&＃xff0c;work工作消息模型

发布订阅&＃xff1a;一个生产者生产的消息&＃xff0c;可以由订阅了消息的所有消费者进行处理

fanout&＃xff0c; direct&＃xff0c; topic

​ &＃xff08;dəˈrekt&＃xff09;&＃xff08;ˈtäpik&＃xff09;

基本消息模型

RabbitMQ是一个消息代理&＃xff1a;它接受和转发消息。 可以把它想象成一个邮局&＃xff1a;当你把邮件放在邮箱里时&＃xff0c;你可以确定邮差最终会把邮件发送给你的收件人。 在这个比喻中&＃xff0c;RabbitMQ是信箱&＃xff0c;邮局和邮递员。RabbitMQ与邮局的主要区别是它不处理纸张&＃xff0c;而是接受&＃xff0c;存储和转发数据消息的二进制数据块。

P&＃xff08; publisher[ˈpəbliSHər]&＃xff09;&＃xff1a;生产者&＃xff0c;一个发送消息的用户应用程序。

C&＃xff08;consumer[kənˈso͞omər]&＃xff09;&＃xff1a;消费者&＃xff0c;消费者是一个用来等待接收消息的用户应用程序

队列&＃xff1a;rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序&＃xff0c;但是它们只能存储在队列中。队列只受主机的内存和磁盘限制&＃xff0c;实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列&＃xff0c;许多消费者可以尝试从一个队列接收数据。

总之&＃xff1a;

生产者将消息发送到队列&＃xff0c;消费者从队列中获取消息&＃xff0c;队列是存储消息的缓冲区。

work消息模型

工作队列或者竞争消费者模式&＃xff1a;能者多劳

工作队列&＃xff0c;又称任务队列。主要思想就是避免执行资源密集型任务时&＃xff0c;必须等待它执行完成。相反我们稍后完成任务&＃xff0c;我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多消费者时&＃xff0c;任务将在他们之间共享&＃xff0c;但是一个消息只能被一个消费者获取。

这个概念在Web应用程序中特别有用&＃xff0c;因为在短的HTTP请求窗口中无法处理复杂的任务。

如何实现能者多劳呢

我们可以使用basicQos方法和prefetchCount &＃61; 1设置。 这告诉RabbitMQ一次不要向工作人员发送多于一条消息。 或者换句话说&＃xff0c;不要向工作人员发送新消息&＃xff0c;直到它处理并确认了前一个消息。 相反&＃xff0c;它会将其分派给不是仍然忙碌的下一个工作人员。

如何避免消息堆积&＃xff1f;

1&＃xff09;采用workqueue&＃xff0c;多个消费者监听同一队列。

2&＃xff09;接收到消息以后&＃xff0c;而是通过线程池&＃xff0c;异步消费。

订阅模型分类

在之前的模式中&＃xff0c;我们创建了一个工作队列。 工作队列背后的假设是&＃xff1a;每个任务只被传递给一个工作人员。 在这一部分&＃xff0c;我们将做一些完全不同的事情 - 我们将会传递一个信息给多个消费者。 这种模式被称为“发布/订阅”。

1、1个生产者&＃xff0c;多个消费者

2、每一个消费者都有自己的一个队列

3、生产者没有将消息直接发送到队列&＃xff0c;而是发送到了交换机

4、每个队列都要绑定到交换机

5、生产者发送的消息&＃xff0c;经过交换机到达队列&＃xff0c;实现一个消息被多个消费者获取的目的

X&＃xff08;Exchanges&＃xff09;&＃xff1a;交换机一方面&＃xff1a;接收生产者发送的消息。另一方面&＃xff1a;知道如何处理消息&＃xff0c;例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作&＃xff0c;取决于Exchange的类型。

Exchange类型有以下几种&＃xff1a;

Fanout&＃xff1a;广播&＃xff0c;将消息交给所有绑定到交换机的队列Direct&＃xff1a;定向&＃xff0c;把消息交给符合指定routing key 的队列 Topic&＃xff1a;通配符&＃xff0c;把消息交给符合routing pattern&＃xff08;路由模式&＃xff09;的队列


Exchange&＃xff08;交换机&＃xff09;只负责转发消息&＃xff0c;不具备存储消息的能力&＃xff0c;因此如果没有任何队列与Exchange绑定&＃xff0c;或者没有符合路由规则的队列&＃xff0c;那么消息会丢失&＃xff01;

订阅模型-Fanout

Fanout&＃xff0c;也称为广播。在广播模式下&＃xff0c;消息发送流程是这样的&＃xff1a;

• 1&＃xff09; 可以有多个消费者
• 2&＃xff09; 每个消费者有自己的queue&＃xff08;队列&＃xff09;
• 3&＃xff09; 每个队列都要绑定到Exchange&＃xff08;交换机&＃xff09;
• 4&＃xff09; 生产者发送的消息&＃xff0c;只能发送到交换机&＃xff0c;交换机来决定要发给哪个队列&＃xff0c;生产者无法决定。
• 5&＃xff09; 交换机把消息发送给绑定过的所有队列
• 6&＃xff09; 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

订阅模型-Direct

在Direct模型下&＃xff0c;队列与交换机的绑定&＃xff0c;不能是任意绑定了&＃xff0c;而是要指定一个RoutingKey&＃xff08;路由key&＃xff09;消息的发送方在向Exchange发送消息时&＃xff0c;也必须指定消息的routing key

订阅模型-Topic

Topic类型的Exchange与Direct相比&＃xff0c;都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符&＃xff01;

持久化

如果我们希望即使在RabbitMQ服务重启的情况下&＃xff0c;也不会丢失消息&＃xff0c;我们可以将Queue与Message都设置为可持久化的&＃xff08;durable&＃xff09;&＃xff0c;这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。当然还是会有一些小概率事件会导致消息丢失

如何避免消息丢失&＃xff1f;

1&＃xff09; 消费者的ACK机制。可以防止消费者丢失消息。

​ 什么是消息确认ACK。

如果在处理消息的过程中&＃xff0c;消费者的服务器在处理消息的时候出现异常&＃xff0c;那么可能这条正在处理的消息就没有完成消息消费&＃xff0c;数据就会丢失。为了确保数据不会丢失&＃xff0c;RabbitMQ支持消息确定-ACK。


​ ACK的消息确认机制

ACK机制是消费者从RabbitMQ收到消息并处理完成后&＃xff0c;反馈给RabbitMQ&＃xff0c;RabbitMQ收到反馈后才将此消息从队列中删除。(注&＃xff1a;息永远不会从RabbitMQ中删除&＃xff0c;只有当消费者正确发送ACK反馈&＃xff0c;RabbitMQ确认收到后&＃xff0c;消息才会从RabbitMQ服务器的数据中删除。)如果一个消费者在处理消息出现了网络不稳定、服务器异常等现象&＃xff0c;那么就不会有ACK反馈&＃xff0c;RabbitMQ会认为这个消息没有正常消费&＃xff0c;会将消息重新放入队列中。如果在集群的情况下&＃xff0c;RabbitMQ会立即将这个消息推送给这个在线的其他消费者。这种机制保证了在消费者服务端故障的时候&＃xff0c;不丢失任何消息和任务。消息的ACK确认机制默认是打开的。


​ ACK机制的开发注意事项

如果忘记了ACK&＃xff0c;那么后果很严重。当Consumer退出时候&＃xff0c;Message会一直重新分发。然后RabbitMQ会占用越来越多的内容&＃xff0c;由于RabbitMQ会长时间运行&＃xff0c;因此这个"内存泄漏"是致命的。


2&＃xff09; 但是&＃xff0c;如果在消费者消费之前&＃xff0c;MQ就宕机了&＃xff0c;消息就没了。

项目过程

思路分析

发送方&＃xff1a;商品微服务

• 什么时候发&＃xff1f;

  当商品服务对商品进行写操作&＃xff1a;增、删、改的时候&＃xff0c;需要发送一条消息&＃xff0c;通知其它服务。

• 发送什么内容&＃xff1f;

  对商品的增删改时其它服务可能需要新的商品数据&＃xff0c;但是如果消息内容中包含全部商品信息&＃xff0c;数据量太大&＃xff0c;而且并不是每个服务都需要全部的信息。因此我们只发送商品id&＃xff0c;其它服务可以根据id查询自己需要的信息。

接收方&＃xff1a;搜索微服务、静态页微服务

接收消息后如何处理&＃xff1f;

• 搜索微服务&＃xff1a;
  • 增/改&＃xff1a;添加新的数据到索引库
  • 删&＃xff1a;删除索引库数据
• 静态页微服务&＃xff1a;
  • 增/改&＃xff1a;创建新的静态页
  • 删&＃xff1a;删除原来的静态页

实现

在pom引入依赖

我们在application.yml中添加一些有关RabbitMQ的配置

在GoodsService中封装一个发送消息到mq的方法

单独编写一个 监听器 处理insert和update、delete的消息

关于RabbitMQ分布式的和工作原理的一些拓展

分布式系统

我们先假设一下&＃xff0c;设想我们正在做一个电商网站。有用户下订单了&＃xff0c;除了要在数据库中新建一条记录之外&＃xff0c;很显然我们还需要给用户发送一封邮件通知他们订单的详情以及一份报表&＃xff0c;以便将来某些时候用户可能会用到。

如果画流程图的话&＃xff0c;可能会是这样&＃xff1a;

但是&＃xff0c;上述这个解决方案的问题在于发送邮件和生成报表都是非常耗时的任务。如果我们在一个请求/响应周期当中&＃xff0c;使用同一个进程来处理这2个耗时任务&＃xff0c;那么用户将会从服务器端等待比较长的时间。甚至&＃xff0c;你的应用服务将会变得很难扩展&＃xff0c;因为越多用户向服务器发起请求&＃xff0c;就要花越多的时间去处理这些请求。而且&＃xff0c;一旦在处理请求上花费很多时间&＃xff0c;那么就会给服务器增加负担&＃xff0c;更坏的情况下&＃xff0c;如果服务器处理的时间很长&＃xff0c;那么服务器甚至会向用户返回一个请求超时的错误。

解决办法是让Web应用解耦。Web应用可以首先将消息发送给消息代理(message broker)&＃xff0c;然后由消息代理将这些消息分发给能执行这些任务的消息消费者&＃xff0c;这样一来&＃xff0c;Web应用就不用亲自去执行这些任务了。

基本上&＃xff0c;消费者是相互之间能独立分开工作的程序&＃xff0c;并且一般情况下消费者程序来自web应用本身。而那些用来服务消费者的服务器&＃xff0c;可以坐落在不同的地方。

除了能减轻服务器的压力之外&＃xff0c;分布式系统的另外一个优势是即使其中一个应用挂了&＃xff0c;整套系统仍然还是可以工作的。假如其中一个消费者无法给用户发送通知邮件了&＃xff0c;那么我们可以把它停掉。即使我们的消费者挂了&＃xff0c;我们的web应用仍然可以继续处理用户的请求并且给代理发送消息。一旦消费者恢复了&＃xff0c;它马上就能接收到之前web应用发来的消息。

现在我们来看一下RabbitMQ ,它是一个在生产者(Web应用)和消费者之间的中间人。

RabbitMQ 要点知识

RabbitMQ 是一个消息代理。它实现了不同的协议&＃xff0c;但是最重要的是&＃xff0c;它实现了AMQP(高级消息队列协议)&＃xff0c;AMQP是一个用来在多个系统之间通过生产者&＃xff0c;代理以及消费者交换消息的协议。

AMQP是如何工作的

现在&＃xff0c;我们有一个生产者和一个消费者。生产者产生消息&＃xff0c;消费者消费消息。在它们二者之间我们还有一个代理&＃xff0c;代理从生产者那里接收消息然后发送给消费者。

如果我们仔细研究下代理的工作原理&＃xff0c;可能会有些难理解。代理由如下3个组件组成&＃xff1a;

1. Exchange – 接受生产者发送的消息&＃xff0c;并将消息路由给Queue。
2. Queue – 消息队列&＃xff0c;一种磁盘或者内存中用来存储消息的数据结构&＃xff1b;
3. Binding – 连接Exchange和Queue&＃xff0c;它告诉Exchange消息应该被传送到哪个Queue。

当创建Exchange时&＃xff0c;我们会指定一个exchange类型。当创建一个binding用来连接一个exchange和一个队列时&＃xff0c;我们会指定一个Binding key。 当发布一条消息时&＃xff0c;我们会指定一个exchange和一个routing Key。 哪条消息会被发送给哪个queue&＃xff0c;取决于下面这4个标准&＃xff1a;

一共有4种类型的exchange:

• Fanout. 这种类型的exchange只是简单地发送消息给它知道的所有队列。

• Direct. 这种类型的exchange会发送消息给符合routing key &＃61; binding key条件的队列。

• Topic. 这种类型的exchange发送消息给符合routing key能部分匹配binding key的队列。

• Header. 这种类型的exchange允许你跟你根据header的值来路由消息&＃xff0c;而不是根据routing key。

最后&＃xff0c;说点题外话&＃xff0c;默认情况下RabbitMQ其实是有一个匿名的exchange。这个exchange会用队列的名字跟routing key做匹配&＃xff0c;而不是binding key。所以&＃xff0c;如果你发布一个routing key &＃61; “order”的消息到这个exchange&＃xff0c;exchange将会路由这个消息到名为“order”的队列。

RabbitMQ三种分布式策略总结

RabbitMQ可以通过三种方法部署分布式集群策略&＃xff1a;Cluster集群、联盟&＃xff08;federation&＃xff09;和shovel。

一 Cluster集群

• 通过连接多个队列服务器节点组成的队列服务器集群。
• 服务器节点之间通信要借助于Erlang的消息传输&＃xff0c;要求集群中所有节点必须有相同的Erlang COOKIE
• 服务器节点之间网络必须是可靠的&＃xff0c;且运行相同版本的RabbitMQ和Erlang。
• 服务器节点之间共享虚拟主机、交换机、用户信息和权限信息。队列可能位于单个节点或镜像到多个节点。连接到任意节点的客户端能够看到集群中所有队列&＃xff0c;即使该队列不位于连接节点上。
• 不支持跨网段&＃xff0c;用于同一个网段内的局域网
  通常可以使用Cluster集群来提高队列可靠性和吞吐量

Cluster集群常用配置方式

至少包含一个磁盘服务器节点节点&＃xff0c;N个内存节点
集群中有两种节点&＃xff1a;
1 内存节点&＃xff1a;只保存状态到内存&＃xff08;一个例外的情况是&＃xff1a;持久的queue的持久内容将被保存到disk&＃xff09;
2 磁盘节点&＃xff1a;保存状态到内存和磁盘。
内存节点队列服务器数据保存在内存&＃xff0c;宕机会丢失未读取的的数据。内存节点虽然不写入磁盘&＃xff0c;但是它执行比磁盘节点要好。
磁盘节点队列服务器数据保存在硬盘&＃xff0c;宕机不会丢失未读取的消息

二 联盟&＃xff08;federation&＃xff09;集群

联盟模式允许单台服务器上的交换机或队列接收到另一台服务器上交换机或队列的消息&＃xff0c;可以是单独机器或集群。

• 服务器节点之间通过AMQP协议通信&＃xff0c;节点不必有相同的Erlang COOKIE。
• 服务器节点之间可运行不同版本RabbitMQ和Erlang
• 可以应用于广域网。
  通常使用*联盟模式连接internet上的中间服务器&＃xff0c;用作订阅分发消息或工作队列。

三 Shovel

shovel连接方式与联盟&＃xff08;federation&＃xff09;的连接方式类似&＃xff0c;但它工作在更低层次。shovel接受队列上的消息&＃xff0c;转发到另一台服务器上的交换机。
shovel和联盟类似&＃xff0c;但它比联盟提供更多控制。

一些需要注意的地方

• 集群配置&＃xff1a;

一个集群中多个节点共享一份.erlang.COOKIE文件&＃xff1b;若是没有启用RABBITMQ_USE_LONGNAME&＃xff0c;需要在每个节点的hosts文件中指定其他节点的地址&＃xff0c;不然会找不到其他集群中的节点。

• 脑裂&＃xff1a;

RabbitMQ集群对于网络分区的处理和忍受能力不太好&＃xff0c;推荐使用federation或者shovel插件去解决。federation详见高级->Federation。但是&＃xff0c;情况已经发生了&＃xff0c;怎么去解决呢&＃xff1f;放心&＃xff0c;还是有办法恢复的。当网络断断续续时&＃xff0c;会使得节点之间的通信断掉&＃xff0c;进而造成集群被分隔开的情况。这样&＃xff0c;每个小集群之后便只处理各自本地的连接和消息&＃xff0c;从而导致数据不同步。当重新恢复网络连接时&＃xff0c;它们彼此都认为是对方挂了&＃xff0c;便可以判断出有网络分区出现了。但是RabbitMQ默认是忽略掉不处理的&＃xff0c;造成两个节点继续各自为政&＃xff08;路由&＃xff0c;绑定关系&＃xff0c;队列等可以独立地创建删除&＃xff0c;甚至主备队列也会每一方拥有自己的master&＃xff09;。可以更改配置使得连接恢复时&＃xff0c;会根据配置自动恢复。

• ignore&＃xff1a;默认&＃xff0c;不做任何处理
• pause-minority&＃xff1a;断开连接时&＃xff0c;判断当前节点是否属于少数派&＃xff08;节点数少于或者等于一半&＃xff09;&＃xff0c;如果是&＃xff0c;则暂停直到恢复连接。
• {pause_if_all_down, [nodes], ignore | autoheal}&＃xff1a;断开连接时&＃xff0c;判断当前集群中节点是否有节点在nodes中&＃xff0c;如果有&＃xff0c;则继续运行&＃xff0c;否则暂停直到恢复连接。这种策略下&＃xff0c;当恢复连接时&＃xff0c;可能会有多个分区存活&＃xff0c;所以&＃xff0c;最后一个参数决定它们怎么合并。
• autoheal&＃xff1a;当恢复连接时&＃xff0c;选择客户端连接数最多的节点状态为主&＃xff0c;重启其他节点。

配置&＃xff1a;**【详见下文&＃xff1a;集群配置】

• 多次ack&＃xff1a;客户端多次应答同一条消息&＃xff0c;会使得该客户端收不到后续消息。