从 0 到 1 ,手把手教你编写《消息队列》项目(Java实现) —

文章目录

一、消息队列是什么？
二、需求分析
- 结构解析
- 功能解析
- 规则解析
- - 绑定关系
  - 交换机类型
  - 消息应答
三、持久化存储
四、网络通信
- 提供的API
- 复用TCP连接
五、消息队列概念图

一、消息队列是什么？

消息队列 (Message Queue, MQ)就是将阻塞队列这一数据结构提取成了一个独立程序,故消息队列也是一个生产者消费者模型.

生产者消费者模型的作用:

1️⃣解耦合:
如果是客户端A直接调用服务器B,那么客户端A与服务器B的代码耦合度就比较高,更改任意一方的代码,可能另一方的代码都会随之受到牵连而也需要改写.

而如果是客户端A将请求发送至生产者消费者模型,服务器B再从中取出,此时A和B的代码耦合度就低,就不会出现牵一发而动全身这样麻烦的情况了.

2️⃣削峰填谷:
服务器C,业务逻辑比较复杂,无法同时处理很多的请求,如果此时有多个客户端向服务器C发送请求,那么服务器C就有挂掉的风险.
引入消息队列后,让所有的客户端都将请求发送到消息队列中,服务器从消息队列中取,
在客户端请求较多时,仍让服务器收到的请求保持在安全数量,即达到削峰效果.
在服务器较为空闲时,仍可以让其从消息队列中获取请求,达到填谷效果.
以此来降低因请求过多而让服务器挂掉的风险.

使用场景:

双十一的时候，淘宝客户端会产生大量订单,客户端将订单提交到服务器,此时服务器就会面临大量订单,那么服务器面对远超自己承受能力的庞大数据量,可能就会挂掉.

此时就可以引入生产者消费者模型 >>> 消息队列,让淘宝客户端的订单,发送到消息队列,再让服务器根据自己的承受能力自行取订单.

二、需求分析

此处咱们编写的消息队列,主要参考市面上比较知名的消息队列 RabbitMQ.
消息队列的核心功能主要有两个:
1️⃣让生产者将需要转发的数据(称之为消息)存储到消息队列中.
2️⃣让消费者将需要取出的消息,从消息队列中取出
在这里插入图片描述

此处的消息队列会有N多个生产者,N多个消费者,其中Borker Server 最为重要,是用来存储和转发消息的.

根据上述要实现的两个核心功能就提取出了六个核心概念:
1️⃣生产者(Producer):生产消息的一方

2️⃣消费者(Consumer):消费消息的一方

3️⃣中间人(Broker):存放消息的一方

4️⃣发布消息(Publish):生产者将生产出的消息,存放到中间人处的功能.

5️⃣订阅消息(Subscribe):消费者订阅中间人此处的某些消息的功能.

6️⃣消费消息(Consume):服务器将消息推送给订阅的消费者的功能.

结构解析

前三个概念是比较好理解的,那我们就来剖析一下后三个功能.

让我们先来看看RabbitMQ的内部结构

在这里插入图片描述

上述结构遵循了AMPQ协议.

Broker Server内部的结构有多个<虚拟机>.

虚拟机:类似于MySQL中的 batabase,是一组数据的逻辑集合.

交换机:生产者投递消息到 Broker Server 实际上,是把消息投递到某个交换机,再由交换机把自身的规则把消息转发给对应的队列.

消息队列:真正用来存储消息的实体.

Binding绑定关系:记录交换机与队列之间的绑定关系.

功能解析

发布消息功能:其实就是生产者将消息发送给对应的交换机,交换机再根据不同的转发规则,转发给与之相绑定且符合规则的消息队列.

订阅消息功能:其实就是消费者去某个消息队列处注册,表明自己要从这个消息队列中取消息,这个注册称之为订阅.(此处的订阅消息就好像是在抖音上给一个博主点了关注,)

消费消息功能:其实就是当消息队列中有消息时,自动向在这个消息队列注册过的消费者推送信息.(当你关注的博主发布新视频时,便会自动将视频推送给你)

规则解析

上述三个功能中有两个核心规则:
1️⃣让生产者的消息正确存储到对应队列的规则(姑且称为正确存储规则).

2️⃣保证将消息转发到对应的消费者手中的规则,而没有在网络上丢失(姑且称为正确转发规则).

正确存储规则要如何实现呢?
发布消息的本质是将消息到交换机,再由对应的交换机存储到与之绑定且符合转发规则的队列中.
故正确存储规则的实现主要依靠
绑定关系与交换机的类型.

绑定关系

绑定关系本质上是给交换机与消息队列通过一个绑定钥匙进行绑定.
所以绑定关系本质上只有三个字段,
exchangeName(交换机名称),queueName(消息队列名称),bindingKey(绑定钥匙).
在这里插入图片描述

在不同的交换机类型中bindingKey也起到不同的作用.

交换机类型

发送消息到交换机时,一共会提交三个参数,
1.消息,2.交换机名称,3.routingKey
交换机共有四种类型:

Direct类型
Fanout类型
Topic类型
Header类型(其中 Header 这种方式比较复杂, 比较少见. 常用的是前三种交换机类型. 咱们此处也主要实现这三种)

Direct类型(直接交换机): 生产者向该类型交换机发送消息时, 直接指定队列名称,无视绑定关系.(即指定的队列与该交换机有没有绑定都可以)

在这里插入图片描述

Fanout类型(扇出交换机):生产者发送的消息会被发送到所有与该交换机存在绑定关系的队列中.
在这里插入图片描述

Topic类型(主题交换机): 发送消息指定⼀个字符串为 routingKey. 然后去绑定关系表中去查找与该交换机绑定的队列,并且与绑定关系中的 bindingKey匹配成功才发送到这个队列中.(匹配算法后续讲解)

在这里插入图片描述

正确转发规则要如何实现呢?
其实很简单,让消费者拿到消息后,告诉服务器一声就ok了,也就是消息应答.

消息应答

应答模式分成两种.
自动应答 : 消费者只要消费了消息, 就算应答完毕了. Broker 直接删除这个消息.(等于没应答)
手动应答 : 消费者手动调用应答接口, Broker Server 收到应答请求之后, 才真正删除这个消息.

手动应答的目的, 是为了保证消息确实被消费者处理成功了. 在⼀些对于数据可靠性要求高的场景, 比较常见.

三、持久化存储

Exchange交换机, Queue消息队列, Binding绑定关系, Message消息 都有持久化的需求.
以保证当程序重启 / 主机重启, 上述内容不丢失.

这就需要我们将这些数据存储到硬盘中,这样当程序重启/主机重启时,才不会丢失数据.

但是为了保证咱们这个《消息队列》程序能够高效的转发处理数据,所以这些数据也要在内存中存储一份,
并且每次重启后,都要将硬盘的数据读取恢复到内存中.

四、网络通信

⽣产者和消费者都是客户端程序, broker Server 则是作为服务器. 通过网络进⾏通信.
在⽹络通信的过程中, 客户端部分要提供对应的 api, 来实现对服务器的操作.

以生产者创建交换机举例:
在这里插入图片描述

提供的API

创建交换机 (exchangeDeclare)
销毁交换机 (exchangeDelete)
创建队列 (queueDeclare)
销毁队列 (queueDelete)
创建绑定 (queueBind)
解除绑定 (queueUnbind)
发布消息 (basicPublish)
订阅消息 (basicConsume)
确认消息 (basicAck)