消息中间件RabbitMQ & Redis数据库 原理讲解及应用

RabbitMQ

消息中间件概述

消息中间件(消息队列:Message Queue) 是指一种在需要进行网络通信的系统进行通道的建立，数据或文件发送的中间件。消息中间件的一个重要作用是可以跨平台操作，为不同操作系统上的应用软件集成提供便利。

使用消息队列(中间件)的优点

系统解耦

• 未使用MQ时： A系统通过代码产生了一条数据给D系统，D系统不需要这条该数据，需要在A系统删除访问F系统的代码 又来一个E系统，A系统需要新增调用E的代码，同时A系统设计者还要考虑，当其他系统宕机了，是否要重发还是取消
• 使用MQ后 如果使用MQ，A系统产生一条数据，直接发送到MQ中去，哪个系统需要这条数据自己去MQ里面消费。如果新系统需要这条数据，直接去MQ里消费即可；如果某个系统不需要这条数据，就取消MQ的消费即可。通过这种方式，A系统不用考虑这些问题。

异步通信

如果使用MQ，那么 A 系统连续发送 3 条消息到 MQ 队列中，假如耗时 5ms，A 系统从接受一个请求到返回响应给用户，总时长是 3 + 5 +20= 28ms,用户体验提升,注意4,5,6,7为异步操作

削峰

如果使用 MQ，每秒 5k 个请求写入 MQ(相当于缓存了请求)，A 系统从MQ中慢慢拉取(例如每秒2000个请求),保证小于等于最大处理请求量,从而保证A系统稳定运行

消息队列的缺点

• 系统可用性降低 - 系统引入的外部依赖越多，后期的维护成本加大,虽然系统与系统之间进行了解耦,但是别忘了所有的系统都与MQ建立了联系,一旦MQ部署的服务器宕机,导致所有的系统无法正常通信.
• 系统复杂度提高 - 加入MQ后,还要考虑,消息丢失,消息顺序,消息重复消费等问题
• 数据一致性问题 由于使异步响应,如果B,C,D中其中一个写库失败,而A处理完直接响应成功,用户得到的是成功的结果,而后台整体操作其实是失败

常见的消息中间件产品

• RabbitMQ（Java开发） 支持多种消息协议，如AMQP、STOMP、MQTT等。具有灵活的路由机制、可靠性的消息传递和高可用性
• ActiveMQ 成熟稳定、支持多种编程语言和协议，如JMS,AMQP等。具有灵活的配置和部署方式，可以在不同的环境中使用
• RocketMQ（电商开发） 高吞吐量、低延迟、可靠性高。支持事务消息、顺序消息等特性，适用于大规模分布式系统
• Kafka（常用于大数据开发、Java开发） 高吞吐量，可扩展性强，分布式架构。能够处理大量的实时数据，支持分区和副本机制，确保数据的可靠性和高可用性

RabbitMQ概述

• AMQP:高级消息队列协议(Advanced Message Queuing Protocol)是面向消息中间件提供的开放的应用层协议，其设计目标是对于消息的排序,路由（包括点对点和订阅-发布),保持可靠性、保证安全性
• RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，专门为开发高并发和分布式系统的一种语言。

RabbitMQ原理图

• Broker：简单来说就是消息队列服务器实体。
• Producer：消息生产者，就是投递消息的程序。
• Consumer：消息消费者，就是接受消息的程序。
• Connection:一个网络连接，比如TCP/IP套接字连接。
• Channel：消息通道，是建立在真实的TCP连接内的虚拟连接（是我们与RabbitMQ打交道的最重要的一个接口）。。仅仅创建了客户端到Broker之间的连接后，客户端还是不能发送消息的,需要为每一个Connection创建Channel，AMQP协议规定只有通过Channel才能执行AMQP的命令。 - 之所以需要Channel，是因为TCP连接的建立和释放都是十分昂贵的，如果一个客户端每一个线程都需要与Broker交互，如果每一个线程都建立一个TCP连接，暂且不考虑TCP连接是否浪费，就算操作系统也无法承受每秒建立如此多的TCP连接。- 建议客户端线程之间不要共用Channel，至少要保证共用Channel的线程发送消息必须是串行的，但是建议尽量共用Connection。
• Exchange：消息交换机，生产者不是直接将消息投递到Queue中的，实际上是生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。
• Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。
• Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
• vhost：虚拟主机，一个broker里可以开设多个vhost，用作权限分离，把不同的系统使用的rabbitmq区分开，共用一个消息队列服务器，但看上去就像各自在用不用的rabbitmq服务器一样。

访问RabbitMQ控制台

• 用户名和密码均为guest

控制台添加用户

RabbitMQ主要工作模式

• simple(简单模式)
• work queues(工作模式)
• publish/subscribe(发布订阅模式)
• routing(路由模式)
• topics(主题模式)
• RPC(远程过程调用)
• Publisher Confirms (发布者确认)

simple模式

一对一：一个生产者，一个消费者

work queues模式

一对多：一个生产者对应多个消费者，但是只能有一个消费者获得消息

publish/subscribe模式

一对多：一个生产者发送的消息会被多个消费者获取，一个生产者、一个交换机、多个队列、多个消费者
Exchange：交换机(X)。一方面，接收消息；另一方面：知道如何处理消息，指定交换机即可
常见一下3种类型：

• Fanout：广播，将消息交给所有绑定到交换机的队列
• Direct：定向，把消息交给符合指定routing key 的队列，配合routing模式使用
• Topic：通配符，将消息交给符合制定规则的队列，配合topics模式使用
• 其他：header：会将消息交给匹配头部参数的队列上

Fanout交换机

扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列。
因为广播不需要“思考”，所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。

routing模式

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个 routingKey
• 消息的发送方在向 Exchange 发送消息时，也必须指定消息的 routingKey

Direct交换机

exchange 不再把消息交给每一个绑定的队列，而是根据消息的 routing key 进行判断，只有队列的routing key 与消息的 routing key 完全一致，才会接收到消息

topics模式

• *:匹配一个单词 # :匹配0或多个单词
• 发送到主题交换机上的message需要携带指定规则的routingKey，主题交换机会根据这个规则将数据发送到对应的(多个)队列上。

Topic交换机

如何保证消息的可靠性

我们根据消息的传递过程会发现可能存在以下问题:

• 消息从生产者发送到 RabbitMQ， 生产者把消息发到 Broker 中的Excahnge之后，如何知道自己的消息有没有被 exchange 成功接收？
• 消息从 Exchange 到 Queue， Exchange 会绑定相关的队列，那么如何知道消息是否正确分发到相应的队列?
• Broker 怎么知道消费者已经接收了消息呢？

此时需要采取手段来保证消息在投递,传送和消费整个过程中的可靠性,针对每个环节进行可靠性保证

producer到Exchange的消息确认

实现方法：
在channel上绑定一个确认机制
底层相当于在channel上绑定一个"confirm Listener"的监听器，producer到exchange
无论是否成功都会执行回调

Exchange到Queue的消息确认

返回机制(回退机制):Exchange->Queue

• 底层相当于在channel上绑定了一个"return listener"的监听器,监听exchange路由到queue失败状态
• mandatory参数设置为true(在application.yml中设置),Exhcange无法路由到Queue,那么会执行回调函数,并传递消息给回调函数
• 如果为设置为false(默认值),broker(RabbitMQ)会直接删除该消息并且不会执行回调

Broker到Consumer的ACK机制

ACK指acknowledge(确认)机制，表示消费者消费者消费后的确认机制，常用的确认的方式有两种

• 自动确认机制 当消息一旦被Consumer接收到，则自动确认收到，RabbitMQ收到应答后会将消息从队列中移除
• 手动确认机制 如果设置了手动确认方式,则需要在业务处理成功后，调用方法来手动确认,如果出现异常，则调用方法，让其自动重新发送消息。

保证消息的幂等性

消息幂等性，其实就是保证同一个消息不被消费者重复消费两次。当消费者消费完消息之后，通常会发送一个ack应答确认信息给生产者，但是这中间有可能因为网络中断等原因，导致broker未能收到确认消息，由此这条消息将会被重复发送给其他消费者进行消费，最终导致相同消息被重复消费

• 解决方案: - 1.生产者在发送消息时,给消息绑定一个唯一ID- 2.消费者开始消费前，根据唯一ID去redis中查询 如果没有,直接消费,并把 id=消息内容 键值对形式写入redis (入库) 如果有,说明已经消费过,不再重复消费

死信交换机和死信队列

概述

• 称为死信队列的必要条件 - 在存储该消息的时候超过队列容纳的最大条数- 消息被拒绝（channel.basicNack()/channel.basicReject(),都要设置requeue=false(不再发回原队列);- 消息过期

//定义死信交换机publicExchangedlExchange(){returnExchangeBuilder.topicExchange(DL_EXCHANGE_NAME).durable(true).build();}//将正常队列绑定到死信交换机
params.put("x-dead-letter-exchange",EXCHANGE_DL_NAME);

实现延时队列

概述

延时队列，最重要的特性就体现在它的延时属性上，跟普通的队列不一样的是，普通队列中的元素总是等着希望被早点取出处理，而延时队列中的元素则是希望被在指定时间得到取出和处理，所以延时队列中的元素是都是带时间属性的，通常来说是需要被处理的消息或者任务,可以简单理解为延迟队列中的消息不会立即被消费,而是到了设定的时间后,才会被消费

• 延时队列的应用场景 - 订单在24h之内未支付则自动取消。- 用户注册成功后，如果三天内没有登陆则进行短信提醒- 预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会

通过死信交换机实现延时队列

• 模拟三十分钟后自动取消订单 首先，设置ttl=30min，在30min过后消息过期生成死信 通过死信交换机和死信队列生成延时队列，让客户端直接访问死信队列，30min后把消息消费了，就有两种情况： 一种是用户在三十分钟内已经支付订单了 一种是三十分钟内用户并未支付，那就修改订单状态未已取消，并且回滚数据库中的该商品库存

Redis

概述

Redis是用C语言开发的一个开源的高性能键值对（key-value）数据库，官方提供测试数据，50个并发执行100000个请求,读的速度是110000次/s,写的速度是81000次/s ，且Redis通过提供多种键值数据类型来适应不同场景下的存储需求，目前为止Redis支持的键值数据类型如下：

• 字符串类型 string
• 散列类型 hash
• 列表类型 list
• 集合类型 set
• 有序集合类型 sortedset

应用场景

• 缓存（数据查询、商品库存等等）
• 任务队列。（商品秒杀、抢购、12306等等）
• 网站访问统计
• 数据过期处理（可以精确到毫秒)
• 分布式集群架构中的session分离

Redis的数据类型

redis是一种高级的key-value的存储系统，其中value支持五种数据类型：

• 字符串类型 string
• 散列类型 hash
• 列表类型 list
• 集合类型 set
• 有序集合类型 sortedset

Jedis的基本使用

Redis不仅是使用命令来操作，现在基本上主流的语言都有客户端支持，比如java、C、C#、C++、php、Node.js、Go等。 在官方网站里列一些Java的客户端，有Jedis、Redisson、Jredis、JDBCRedis、等其中官方推荐使用Jedis和Redisson。java操作redis的第三方类库: jedis

Jedis连接池

jedis连接资源的创建与销毁是很消耗程序性能，所以jedis为我们提供了jedis的池化技术，jedisPool在创建时初始化一些连接资源存储到连接池中，使用jedis连接资源时不需要创建，而是从连接池中获取一个资源进行redis的操作，使用完毕后，不需要销毁该jedis连接资源，而是将该资源归还给连接池，供其他请求使用。