消息队列基础知识

学一点，整一点，基本都是综合别人的，弄成我能理解的内容
https://blog.csdn.net/BenJamin_Blue/article/details/125946812
https://blog.csdn.net/qq_46119575/article/details/129794304

消息队列（Message Queue，MQ)是一种进程间通信或同一进程的不同线程间的通信方式。被广泛应用为分布式服务框架的消息中间件。消息发送后可以立即返回，由消息系统来确保消息的可靠传递。消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的。这样发布者和使用者都不用知道对方的存在。

生产者-消费者模型

由生产者发布消息队列至消息服务器，再由消费者订阅消息。

生产者（Producer）业务的发起方，负责生产消息发布给Broker。
消费者（Consumer）业务的处理方，负责从Broker订阅消息并进行业务逻辑处理。
消息服务器（Broker）MQ的服务器。包括接收 Producer 发过来的消息、处理 Consumer 的消费消息请求、消息的持久化存储、以及服务端过滤功能等。

注意消息服务器可以是，分布式，或多节点的集群，且每个节点里可能不止一个队列。

当然除了消息服务器外，生产者、消费者和消息本身也可以拥有集群的概念。
我们可以对这三者进行分组，形成主题的概念，并进一步细化出二级标签，实现特定的集群收发特定消息的功能。

主题（topic）一级消息类型，不同生产者向特定的topic发送消息，再由MQ分发至特定的订阅者，实现消息的传递。
标签（tag）二级消息类型，用来进一步区分某个Topic下的消息子类。
集群（group）一组生产者或消费者，这组生产者或消费者通常生产或消费同一类消息，且消息发布或订阅的逻辑一致。

弊端：
1、消息的收发依赖于中间件，且中间件的稳定运行需要维护成本。
2、提高开发复杂度。需要考虑消息的处理，包括消息幂等性（重复消费问题）、消息中间件的持久化和稳定性、可靠性等。

MQ持久化：MQ会将你的持久化消息写入磁盘上的持久化日志文件，等消息被消费之后，RabbitMQ会把这条消息标识为等待垃圾回收。
缺点：性能低，写入硬盘要比写入内存性能较低很多，从而降低了服务器的吞吐量。

消息幂等性

重复消费：生产者多发、消费者多次消费等。
解决这种问题的常用办法，就是保证操作的幂等性：
幂等操作（Idempotent Operation）：执行任意多次幂等操作所产生的影响均与一次执行的效果相同。

举个例子，数据库脚本insert前都会先delete，这么一组数据库操作无论执行多少次结果都是一样的。
重复消费的问题也可以用这个思想去解决。

实现：
1、消息中间件端根据 Message Id 去重。
2、消费端： 数据库：新增/修改。 组件如redis进行自身去重。

Kafka 实际上有个 offset 的概念，就是每个消息写进去，都有一个 offset，代表消息的序号，然后 consumer 消费了数据之后，每隔一段时间（定时定期），会把自己消费过的消息的 offset 提交一下，表示“我已经消费过了，下次我要是重启啥的，你就让我继续从上次消费到的 offset 来继续消费吧”。

但是凡事总有意外，你有时候重启系统，看你怎么重启了，如果碰到点着急的，直接 kill 进程了，再重启。这会导致 consumer 有些消息处理了，但是没来得及提交 offset，尴尬了。重启之后，少数消息会再次消费一次。

场景：数据 1/2/3 依次进入 kafka，kafka 会给这三条数据每条分配一个 offset，代表这条数据的序号，我们就假设分配的 offset 依次是 152/153/154。消费者从 kafka 去消费的时候，也是按照这个顺序去消费。假如当消费者消费了 offset=153 的这条数据，刚准备去提交 offset 到 zookeeper，此时消费者进程被重启了。那么此时消费过的数据 1/2 的 offset并没有提交，kafka 也就不知道你已经消费了 offset=153 这条数据。那么重启之后，消费者会找kafka 说，嘿，哥儿们，你给我接着把上次我消费到的那个地方后面的数据继续给我传递过来。由于之前的 offset 没有提交成功，那么数据 1/2 会再次传过来，如果此时消费者没有去重的话，那么就会导致重复消费。

如果消费者干的事儿是拿一条数据就往数据库里写一条，会导致说，你可能就把数据 1/2 在数据库里插入了 2 次，那么数据就错啦。
其实重复消费不可怕，可怕的是你没考虑到重复消费之后，怎么保证幂等性。

（1）比如你拿个数据要写库，你先根据主键查一下，如果这数据都有了，你就别插入了，update 一下好吧。
（2）比如你是写 Redis，那没问题了，反正每次都是 set，天然幂等性。
（3）比如你不是上面两个场景，那做的稍微复杂一点，你需要让生产者发送每条数据的时候，里面加一个全局唯一的 id，类似订单 id 之类的东西，然后你这里消费到了之后，先根据这个 id 去比如 Redis 里查一下，之前消费过吗？如果没有消费过，你就处理，然后这个 id 写 Redis。如果消费过了，那你就别处理了，保证别重复处理相同的消息即可。
（4）比如基于数据库的唯一键来保证重复数据不会重复插入多条。因为有唯一键约束了，重复数据插入只会报错，不会导致数据库中出现脏数据

主流MQ

在流量和大数据的时代，ActiveMQ和RabbitMQ这两者因为吞吐量以及GitHub的社区活跃度的原因，在各大互联网公司基本上销声匿迹了，越来越多的公司开始青睐于后两者。其中RocketMQ是阿里开源的，这和同样是阿里开源的rpc框架-dubbo设计风格比较类似。Kafka则更多应用在大数据业务场景中。

现在主要是rocketmq和kafka

消息队列的应用

消息队列常见的有三方面的好处：解耦，异步，削峰

1、业务解耦

A 系统发送数据到 BCD 三个系统，通过接口调用发送。如果 E 系统也要这个数据呢？那如果 D 系统现在不需要了呢？

A 系统跟其它各种乱七八糟的系统严重耦合，A 系统产生一条比较关键的数据，很多系统都需要 A 系统将这个数据发送过来。A 系统要时时刻刻考虑 BCDE 四个系统如果挂了该咋办？要不要重发，要不要把消息存起来？

从这个场景可以看出，一旦发生改动，这种一对一连接的方式会非常麻烦，系统的耦合关系很强，需要改动多个系统的代码。

具体用法：使用 MQ，A 系统产生一条数据，发送到 MQ 里面去，哪个系统需要数据自己去 MQ 里面消费。如果新系统需要数据，直接从 MQ 里消费即可；如果某个系统不需要这条数据了，就取消对 MQ 消息的消费即可。这样下来，A 系统压根儿不需要去考虑要给谁发送数据，不需要维护这个代码，也不需要考虑人家是否调用成功、失败超时等情况。

通过一个 MQ，Pub/Sub 发布订阅消息这么一个模型，A 系统就跟其它系统彻底解耦了

2、异步

场景：A 系统接收一个请求，需要在自己本地写库，还需要在 BCD 三个系统写库，自己本地写库要 3ms，BCD 三个系统分别写库要 300ms、450ms、200ms。最终请求总延时是 3+ 300 + 450 + 200 = 953ms

使用 MQ，那么 A 系统连续发送 3 条消息到 MQ 队列中，假如耗时 5ms，A 系统从接受一个请求到返回响应给用户，总时长是 3 + 5 = 8ms，对于用户而言，其实感觉上就是点个按钮，8ms 以后就直接返回了，真快！
从953ms到8ms，性能提升100多倍，这就是MQ机制带来的好处

这也是为什么平时我们上传一个文件或者发表一个内容，本地显示上传成功或者发表成功了，但是别的用户在短时间内还看不到，需要过一会才能刷新出来。

这其实就是因为数据的同步需要时间，本地看到的上传成功其实是我们把数据成功上传到 MQ 中了，但是 MQ 写入库还需要一定的时间（目前这种延迟已经很短了），所以别的用户暂时还看不到，需要等 MQ 中的数据真正入库完成其它用户才能看到内容。

3、削峰

场景：每天 0:00 到 12:00，A 系统风平浪静，每秒并发请求数量就 50 个。结果每次一到 12:00 ~ 13:00，每秒并发请求数量突然会暴增到 5k+ 条。但是系统是直接基于 MySQL 的，大量的请求涌入MySQL，每秒钟对 MySQL 执行约 5k 条 SQL。

使用 MQ，每秒 5k 个请求写入 MQ，A 系统每秒钟最多处理 2k 个请求，因为 MySQL 每秒钟最多处理 2k 个。
A 系统从 MQ 中慢慢拉取请求，每秒钟就拉取 2k 个请求，不要超过自己每秒能处理的最大请求数量就 ok，这样下来，哪怕是高峰期的时候，A 系统也绝对不会挂掉。
而 MQ 每秒钟 5k 个请求进来，就 2k 个请求出去，结果就导致在中午高峰期（1 个小时），可能有几十万甚至几百万的请求积压在 MQ 中。