小茵：今天要不来聊聊消息队列吧？我看你项目不少地方都写到Kafka了
小奥：嗯嗯
小茵：那你简单说明下你使用Kafka的场景吧
小奥：使用消息队列的目的总的来说可以有三种情况：解耦、异步和削峰
小奥：比如举我项目的例子吧，我现在维护一个消息管理平台系统，对外提供接口给各个业务方调用
小奥：他们调用接口之后，实际上『不是同步』下发了消息。
小奥：在接口处理层只是把该条消息放到了消息队列上，随后就直接返回结果给接口调用者了。
小奥：这样的好处就是：
小奥：1. 接口的吞吐量会大幅度提高（因为未做真正实际调用，接口RT会非常低）【异步】
小奥：2. 即便有大批量的消息调用接口都不会让系统受到影响（流量由消息队列承载）【削峰】
小奥：又比如说，我这边还有个项目是广告订单归因工程，主要做的事情就是得到订单数据，给各个业务广告计算对应的佣金。
小奥：订单的数据是从消息队列里取出的
小奥：这样设计的好处就是：
小奥：1. 交易团队的同学只要把订单消息写到消息队列，该订单数据的Topic由各个业务方自行消费使用【解耦】【异步】
小奥：2. 即便下单QPS猛增，对下游业务无太大的感知（因为下游业务只消费消息队列的数据，不会直接影响到机器性能）【削峰】
小茵：嗯，那我想问下，你觉得为什么消息队列能到削峰？
小茵：或者换个问法，为什么Kafka能承载这么大的QPS？
小奥：消息队列「最核心」的功能就是把生产的数据存储起来，然后给各个业务把数据再读取出来。
小奥：跟我们处理请求时不一样：在业务处理时可能会调别人的接口，可能会需要去查数据库…等等等一系列的操作才行
小奥：像Kafka在「存储」和「读取」这个过程中又做了很多的优化
小奥：举几个例子，比如说：
小奥：我们往一个Topic发送消息或者读取消息时，实际内部是多个Partition在处理【并行】
小奥：在存储消息时，Kafka内部是顺序写磁盘的，并且利用了操作系统的缓冲区来提高性能【append+cache】
小奥：在读写数据中也减少CPU拷贝文件的次数【零拷贝】
小茵：嗯，你既然提到减少CPU拷贝文件的次数，可以给我说下这项技术吗？
小奥：嗯，可以的，其实就是零拷贝技术。
小奥：比如我们正常调用read函数时，会发生以下的步骤:
小奥：1. DMA把磁盘的拷贝到读内核缓存区
小奥：2. CPU把读内核缓冲区的数据拷贝到用户空间
小奥：正常调用write函数时，会发生以下的步骤：
小奥：1. CPU把用户空间的数据拷贝到Socket内核缓存区
小奥：2. DMA把Socket内核缓冲区的数据拷贝到网卡
小奥：可以发现完成「一次读写」需要2次DMA拷贝，2次CPU拷贝。而DMA拷贝是省不了的，所谓的零拷贝技术就是把CPU的拷贝给省掉
小奥：并且为了避免用户进程直接操作内核，保证内核安全，应用程序在调用系统函数时，会发生上下文切换（上述的过程一共会发生4次）
小奥：目前零拷贝技术主要有：mmap和sendfile，这两种技术会一定程度下减少上下文切换和CPU的拷贝
小奥：比如说：mmap是将读缓冲区的地址和用户空间的地址进行映射，实现读内核缓冲区和应用缓冲区共享
小奥：从而减少了从读缓冲区到用户缓冲区的一次CPU拷贝
小奥：使用mmap的后一次读写就可以简化为：
小奥：一、DMA把硬盘数据拷贝到读内核缓冲区。
小奥：二、CPU把读内核缓存区拷贝至Socket内核缓冲区。
小奥：三、DMA把Socket内核缓冲区拷贝至网卡
小奥：由于读内核缓冲区与用户空间做了映射，所以会省了一次CPU拷贝
小奥：而sendfile+DMA Scatter/Gather则是把读内核缓存区的文件描述符/长度信息发到Socket内核缓冲区，实现CPU零拷贝
小奥：使用sendfile+DMA Scatter/Gather一次读写就可以简化为：
小奥：一、DMA把硬盘数据拷贝至读内核缓冲区。
小奥：二、CPU把读缓冲区的文件描述符和长度信息发到Socket缓冲区。
小奥：三、DMA根据文件描述符和数据长度从读内核缓冲区把数据拷贝至网卡
小奥：回到Kafka上
小奥：从Producer->Broker，Kafka是把网卡的数据持久化硬盘，用的是mmap（从2次CPU拷贝减至1次）
小奥：从Broker->Consumer，Kafka是从硬盘的数据发送至网卡，用的是sendFile（实现CPU零拷贝）
小茵：我稍微打断下，我还有点事忙，我总结下你说的话吧
小茵：你用Kafka的原因是为了异步、削峰、解耦
小茵：Kafka能这么快的原因就是实现了并行、充分利用操作系统cache、顺序写和零拷贝
小茵：没错吧？
小奥：嗯
小茵：ok，下次继续面吧， 我这边有点忙
小茵：面试官：今天我想问下，你觉得Kafka会丢数据吗？
小奥：嗯，使用Kafka时，有可能会有以下场景会丢消息
小奥：比如说，我们用Producer发消息至Broker的时候，就有可能会丢消息
小奥：如果你不想丢消息，那在发送消息的时候，需要选择带有 callBack的api进行发送
小奥：其实就意味着，如果你发送成功了，会回调告诉你已经发送成功了。如果失败了，那收到回调之后自己在业务上做重试就好了。
小奥：等到把消息发送到Broker以后，也有可能丢消息
小奥：一般我们的线上环境都是集群环境下嘛，但可能你发送的消息后broker就挂了，这时挂掉的broker还没来得及把数据同步给别的broker，数据就自然就丢了
小奥：发送到Broker之后，也不能保证数据就一定不丢了，毕竟Broker会把数据存储到磁盘之前，走的是操作系统缓存
小奥：也就是异步刷盘这个过程还有可能导致数据会丢
小奥：嗯，到这里其实我已经说了三个场景了，分别是：producer -> broker ，broker->broker之间同步，以及broker->磁盘
小奥：要解决上面所讲的问题也比较简单，这块也没什么好说的…
小奥：不想丢数据，那就使用带有callback的api，设置 acks、retries、factor等等些参数来保证Producer发送的消息不会丢就好啦。
小茵：嗯…
小奥：一般来说，还是client 消费 broker 丢消息的场景比较多
小茵：那你们在消费数据的时候是怎么保证数据的可靠性的呢？
小奥：首先，要想client端消费数据不能丢，肯定是不能使用autoCommit的，所以必须是手动提交的。
小奥：我们这边是这样实现的：
小奥：一、从Kafka拉取消息（一次批量拉取500条，这里主要看配置）时
小奥：二、为每条拉取的消息分配一个msgId（递增）
小奥：三、将msgId存入内存队列（sortSet）中
小奥：四、使用Map存储msgId与msg(有offset相关的信息）的映射关系
小奥：五、当业务处理完消息后，ack时，获取当前处理的消息msgId，然后从sortSet删除该msgId（此时代表已经处理过了）
小奥：六、接着与sortSet队列的首部第一个Id比较（其实就是最小的msgId），如果当前msgId<=sort Set第一个ID，则提交当前offset
小奥：七、系统即便挂了，在下次重启时就会从sortSet队首的消息开始拉取，实现至少处理一次语义
小奥：八、会有少量的消息重复，但只要下游做好幂等就OK了。
小茵：嗯，你也提到了幂等，你们这业务怎么实现幂等性的呢？
小奥：嗯，还是以处理订单消息为例好了。
小奥：幂等Key我们由订单编号+订单状态所组成（一笔订单的状态只会处理一次）
小奥：在处理之前，我们首先会去查Redis是否存在该Key，如果存在，则说明我们已经处理过了，直接丢掉
小奥：如果Redis没处理过，则继续往下处理，最终的逻辑是将处理过的数据插入到业务DB上，再到最后把幂等Key插入到Redis上
小奥：显然，单纯通过Redis是无法保证幂等的（：
小奥：所以，Redis其实只是一个「前置」处理，最终的幂等性是依赖数据库的唯一Key来保证的（唯一Key实际上也是订单编号+状态）
小奥：总的来说，就是通过Redis做前置处理，DB唯一索引做最终保证来实现幂等性的
小茵：你们那边遇到过顺序消费的问题吗？
小奥：嗯，也是有的，我举个例子
小奥：订单的状态比如有 支付、确认收货、完成等等，而订单下还有计费、退款的消息报
小奥：理论上来说，支付的消息报肯定要比退款消息报先到嘛，但程序处理的过程中可不一定的嘛
小奥：所以在这边也是有消费顺序的问题
小奥：但在广告场景下不是「强顺序」的，只要保证最终一致性就好了。
小奥：所以我们这边处理「乱序」消息的实现是这样的：
小奥：一、宽表：将每一个订单状态，单独分出一个或多个独立的字段。消息来时只更新对应的字段就好，消息只会存在短暂的状态不一致问题，但是状态最终是一致的
小奥：二、消息补偿机制：另一个进行消费相同topic的数据，消息落盘，延迟处理。将消息与DB进行对比，如果发现数据不一致，再重新发送消息至主进程处理
小奥：还有部分场景，可能我们只需要把相同userId/orderId发送到相同的partition（因为一个partition由一个Consumer消费），又能解决大部分消费顺序的问题了呢。
小茵：嗯…懂了
