什么是消息队列?
一、什么是消息队列?
消息队列,一般我们会简称它为MQ(Message Queue),嗯,就是很直白的简写。
我们先不管消息(Message)这个词,来看看队列(Queue)。这一看,队列大家应该都熟悉吧。
队列是一种 先进先出 的数据结构。
在Java里边,已经实现了不少的队列了:
那为什么还需要消息队列(MQ)这种 中间件 呢???其实这个问题,跟之前我学Redis的时候很像。Redis是一个以 key-value 形式存储的内存数据库,明明我们可以使用类似HashMap这种实现类就可以达到类似的效果了,那还为什么要Redis?
消息队列可以简单理解为: 把要传输的数据放在队列中 。
二、为什么要用消息队列?
2.1 解耦
现在我有一个系统A,系统A可以产生一个 userId
然后,现在有系统B和系统C都需要这个 userId 去做相关的操作
写成伪代码可能是这样的:
-
Java 代码
01 |
public class SystemA { |
|
02 |
|
03 |
// 系统B和系统C的依赖 |
|
04 |
SystemB systemB = new SystemB(); |
05 |
SystemC systemC = new SystemC(); |
|
06 |
|
07 |
// 系统A独有的数据userId |
|
08 |
private String userId = "Java3y"; |
09 |
|
|
10 |
public void doSomething() { |
11 |
|
|
12 |
// 系统B和系统C都需要拿着系统A的userId去操作其他的事 |
13 |
systemB.SystemBNeed2do(userId); |
14 |
systemC.SystemCNeed2do(userId); |
15 |
|
|
16 |
} |
17 |
} |
结构图如下:
又过了几天,系统D的负责人接了个需求,也需要用到系统A的userId,于是就跑去跟系统A的负责人说:"老哥,我要用到你的userId,你调一下我的接口吧"
于是系统A说:"没问题的,这就搞"
然后,系统A的代码如下:
-
Java 代码
01 |
public class SystemA { |
|
02 |
|
03 |
// 已经不再需要系统B的依赖了 |
|
04 |
// SystemB systemB = new SystemB(); |
05 |
|
|
06 |
// 系统C和系统D的依赖 |
07 |
SystemC systemC = new SystemC(); |
08 |
SystemD systemD = new SystemD(); |
09 |
|
|
10 |
// 系统A独有的数据 |
11 |
private String userId = "Java3y"; |
|
12 |
|
13 |
public void doSomething() { |
|
14 |
|
15 |
|
|
16 |
// 已经不再需要系统B的依赖了 |
17 |
//systemB.SystemBNeed2do(userId); |
|
18 |
|
19 |
// 系统C和系统D都需要拿着系统A的userId去操作其他的事 |
|
20 |
systemC.SystemCNeed2do(userId); |
21 |
systemD.SystemDNeed2do(userId); |
|
22 |
|
23 |
} |
|
24 |
} |
时间飞逝:
• 又过了几天,系统E的负责人过来了,告诉系统A,需要userId。
• 又过了几天,系统B的负责人过来了,告诉系统A,还是重新掉那个接口吧。
• 又过了几天,系统F的负责人过来了,告诉系统A,需要userId。
• …...
于是系统A的负责人,每天都被这给骚扰着,改来改去,改来改去.......
还有另外一个问题,调用系统C的时候,如果系统C挂了,系统A还得想办法处理。如果调用系统D时,由于网络延迟,请求超时了,那系统A是反馈 fail 还是重试??
最后,系统A的负责人,觉得隔一段时间就改来改去,没意思,于是就跑路了。
然后,公司招来一个大佬,大佬经过几天熟悉,上来就说: 将系统A的userId写到消息队列中,这样系统A就不用经常改动了 。为什么呢?下面我们来一起看看:
系统A将userId写到消息队列中,系统C和系统D从消息队列中拿数据。 这样有什么好处 ?
• 系统A 只负责 把数据写到队列中,谁想要或不想要这个数据(消息), 系统A一点都不关心 。
o 即便现在系统D不想要userId这个数据了,系统B又突然想要userId这个数据了,都跟系统A无关,系统A一点代码都不用改。
• 系统D拿userId不再经过系统A,而是从消息队列里边拿。 系统D即便挂了或者请求超时,都跟系统A无关,只跟消息队列有关 。
这样一来,系统A与系统B、C、D都 解耦 了。
2.2 异步
我们再来看看下面这种情况:系统A还是 直接调用 系统B、C、D
代码如下:
-
Java 代码
01 |
public class SystemA { |
|
02 |
|
03 |
SystemB systemB = new SystemB(); |
04 |
SystemC systemC = new SystemC(); |
05 |
SystemD systemD = new SystemD(); |
|
06 |
|
07 |
// 系统A独有的数据 |
|
08 |
private String userId ; |
09 |
|
|
10 |
public void doOrder() { |
11 |
|
|
12 |
// 下订单 |
13 |
userId = this.order(); |
|
14 |
// 如果下单成功,则安排其他系统做一些事 |
15 |
systemB.SystemBNeed2do(userId); |
16 |
systemC.SystemCNeed2do(userId); |
17 |
systemD.SystemDNeed2do(userId); |
|
18 |
|
19 |
} |
|
20 |
} |
假设 系统A运算出userId具体的值需要50ms,调用系统B的接口需要300ms,调用系统C的接口需要300ms,调用系统D的接口需要300ms。那么这次请求就需要 50+300+300+300=950ms
并且我们得知,系统A做的是 主要的业务 ,而系统B、C、D是 非主要 的业务。比如系统A处理的是 订单下单 ,而系统B是订单下单成功了,那发送一条短信告诉具体的用户此订单已成功,而系统C和系统D也是处理一些小事而已。
那么此时,为了 提高用户体验和吞吐量 ,其实可以 异步地 调用系统B、C、D的接口。所以,我们可以弄成是这样的:
系统A执行完了以后,将userId写到消息队列中,然后就直接返回了(至于其他的操作,则异步处理)。
• 本来整个请求需要用950ms(同步)
• 现在将调用其他系统接口异步化,只需要100ms(异步)
2.3削峰/限流
我们再来一个场景,现在我们每个月要搞一次大促,大促期间的并发可能会很高的,比如每秒3000个请求。假设我们现在有两台机器处理请求,并且每台机器只能每次处理1000个请求。
那多出来的1000个请求,可能就把我们 整个系统给搞崩了 ...所以,有一种办法,我们可以写到消息队列中:
系统B和系统C 根据自己的能够处理的请求数去消息队列中拿数据 ,这样即便有每秒有8000个请求,那只是把请求放在消息队列中,去拿消息队列的消息 由系统自己去控制 ,这样就不会把整个系统给搞崩。
2.4 日志处理
日志处理是指将消息队列用在日志处理中,比如Kafka的应用,解决大量日志传输的问题。架构简化如下
• 日志采集客户端,负责日志数据采集,定时写受写入Kafka队列
• Kafka消息队列,负责日志数据的接收,存储和转发
• 日志处理应用:订阅并消费kafka队列中的日志数据
2.5 消息通讯
消息通讯是指,消息队列一般都内置了高效的通信机制,因此也可以用在纯的消息通讯。比如实现点对点消息队列,或者聊天室等
点对点通讯:
聊天室通讯:
客户端A,客户端B,客户端N订阅同一主题,进行消息发布和接收。实现类似聊天室效果。
以上实际是消息队列的两种消息模式,点对点或发布订阅模式。
三、使用消息队列有什么问题?
经过我们上面的场景,我们已经可以发现,消息队列能做的事其实还是蛮多的。
说到这里,我们先回到文章的开头,"明明JDK已经有不少的队列实现了,我们还需要消息队列中间件呢?"其实很简单,JDK实现的队列种类虽然有很多种,但是都是简单的内存队列。为什么我说JDK是简单的内存队列呢?下面我们来看看要实现消息队列(中间件)可能要考虑什么问题。
3.1高可用
无论是我们使用消息队列来做解耦、异步还是削峰,消息队列 肯定不能是单机 的。试着想一下,如果是单机的消息队列,万一这台机器挂了,那我们整个系统几乎就是不可用了。
所以,当我们项目中使用消息队列,都是得 集群 / 分布式 的。要做 集群 / 分布式 就必然希望该消息队列能够提供 现成 的支持,而不是自己写代码手动去实现。
3.2 数据丢失问题
我们将数据写到消息队列上,系统B和C还没来得及取消息队列的数据,就挂掉了。 如果没有做任何的措施,我们的数据就丢了 。
学过Redis的都知道,Redis可以将数据持久化磁盘上,万一Redis挂了,还能从磁盘从将数据恢复过来。同样地,消息队列中的数据也需要存在别的地方,这样才尽可能减少数据的丢失。
那存在哪呢?
• 磁盘?
• 数据库?
• Redis?
• 分布式文件系统?
同步存储还是异步存储?
3.3消费者怎么得到消息队列的数据?
消费者怎么从消息队列里边得到数据?有两种办法:
• 生产者将数据放到消息队列中,消息队列有数据了, 主动叫 消费者去拿(俗称push)
• 消费者不断去 轮训 消息队列,看看有没有新的数据,如果有就消费(俗称pull)
3.4其他
除了这些,我们在 使用的时候 还得考虑各种的问题:
• 消息重复消费了怎么办啊?
• 我想保证消息是 绝对 有顺序的怎么做?
• ……..
虽然消息队列给我们带来了那么多的好处,但同时我们发现引入消息队列也会 提高系统的复杂性 。市面上现在已经有不少消息队列轮子了,每种消息队列都有自己的特点, 选取哪种MQ还得好好斟酌 。
- 点赞
- 收藏
- 关注作者
评论(0)