闲扯kafka mq

本文主要讲解关于kafka mq的设计思想及个人理解。关于kafka的详细信息，大家可以参考官网的文献 http://kafka.apache.org/documentation.html这是一篇相当不错的文章，值得仔细研读。

第一个问题：消息队列（Message Queue）是干嘛用的？

首先，要对消息队列有一个基本的理解。不少人虽然在用消息队列，却并没有搞清楚消息队列是干嘛的。

有人会回答，消息队列就是为了分发消息用的。这当然没错，废话总是真理嘛。那么，消息队列是用来提高性能，加速消息传输的吗？显然不是，消息队列虽然提供了数据上的冗余，但它不是一种缓存。如果你想加速，直接在把生产者与消费者合在一起写，中间自己加一个全内存的queue，没有了持久化，没有了网络传输，岂不更快。有人说，消息队列，就是一个数据源，作为下一级输入的数据源，存放中间结果用的。这当然也没错，但是如果纯作存放中间结果用，你为什么不直接用数据库，或者用redis,说不定性能还更佳。
在我看来，对消息队列最好的诠释，还是之前在看active mq文档时看到的那句:"fire and forget"。说中文，两个字：“解耦”。它实现了生产者与消费者的有效解耦，降低了系统复杂性。作为一个生产者，它主要关心的应该就是自己的生产工作，它不应该关心自己生产的东西，到底被谁消费，如何消费。它应该就是简单的把生产好的东西，往一个仓库一放（即fire），然后就可以不管了（forget）,毫无心理负担。至于后面的事，消息如何交付给消费者，这种交付方式是不是会丢失消息之类的可靠性问题一概不管（这也就是为什么消息队列不仅是一个中间结果存放区的原因）。这个作为中间仓库，负责与消费者打交道，同时保证后续交付可靠性的角色，就是消息队列来担当的。
这里打一个不太和谐的比喻。就好比约炮，开完一炮之后，就转身就走，头都不回，很潇洒，fire and forget。至于后续的事，是不是怀孕了，要奶孩子了，抚养成人之类的问题，producer可以一概不管，由消息队列成功接盘。所以，这里的producer有点类似隔壁老王，而消息队列，则无私担当了冤大头这个伟大角色。

神奇的kafka

相对于传统的jms系统，kafka的设计是相当激进的。传统jms之于kafka，有点类似于mongodb之于mysql，走的是粗犷路线，从一开始的设计上就是追求分布式，高可用与并发性能去的。跟我们老大讨论时，他也提到，active mq是为实现jms去的，所以搞得会过于复杂，而kafka mq根本就不去支持jms,没有约束。
先贴一段，官网上的原话：

The Kafka cluster retains all published messages—whether or not they have been consumed—for a configurable period of time. For example if the log retention is set to two days, then for the two days after a message is published it is available for consumption, after which it will be discarded to free up space. Kafka's performance is effectively constant with respect to data size so retaining lots of data is not a problem.

kafka集群会保存所有发布的消息，无论该消息，是否已经确认被消费者所接收。所有这些消息，是作为log被保存的。 消息存起来，好几天后才删，这一点就很神奇，大部份消息队列在确认consumer已接收之后，很快就会把消息删除（即便是持久化保存的消息）。而更神奇的是，kafka卡的性能基本不会因持久化的信息量的增长而变差，基本为一个常量。
其实这跟kafka的log(即持久化的消息)的存储方式有很大关系，说白了，kafka的log是以数据文件配合索引文件来完成查询的（没错，对kafka的一条消息发送，其实就是一次consumer的一次查询操作），所以每次对通过指定的offset对消息的读取，基本都只需要恒定次数的磁头寻道次数就可以完成。

In fact the only metadata retained on a per-consumer basis is the position of the consumer in the log, called the "offset". This offset is controlled by the consumer: normally a consumer will advance its offset linearly as it reads messages, but in fact the position is controlled by the consumer and it can consume messages in any order it likes. For example a consumer can reset to an older offset to reprocess.

以active mq为例的消息队列，其订阅发布模式，都可以认为是有状态的。消息队列这一头必须要记录consumer的接收情况，然后才能决定，发送哪一条消息。试想一下，就算我们就实现一个简单的数据结构 queue，我们肯定也要记录当前队列的top的引用是指向哪个节点的。众所周知，有状态的服务，难以做横向扩展（直接加机器）。那么，kafka是如何保证其消息发送（其实就是pull查询）是无状态的呢？
从上面的这段官方的英文讲解中可以看出答案，就是kafka这边干脆不记录consumer的具体读取到队列哪个位置的这种状态信息，这个位置信息(也就是offset)，交由每个consumer中负责连接kafka的部分自行管理，例如kafka提供的consumer端的client实现就是将这个offset信息定时存到zookeeper上，而kafka本身所做的事，就快跟一个分布式存储系统差不多了。这样的做法也带来了额外的好处，上面文档中所提的最后一句，一个consumer可以根据一个较早的offset进行查找，重新获得某条消息。估计有人要惊了，这算哪门子的好处，我用来作消息队列，又不是数据库，一般看队列头的消息就够了，为什么老要去查找过去的消息？关于这个问题，下文来表。

分布式kafka

从分布这个角度来看，还是那句话，kafka之于active mq，相当于mongodb之于mysql。无论active mq还是mysql，起始都是从单机开始发展起来的，一开始就不是为了分布式而设计，而后再在原来的基本础上再做分布式的处理。所以这样的分布式，总觉得差那么一点味道，不纯正。例如active mq的Master-Slave模式无法做负载均衡，而Broker Cluster却又不是HA(高可靠)的。 回头看kafka,天生为分布式而生。它的分布式是行列式形式的，如下图。

 每个topic的log信息，被分成多个partition分布在不同的broker（kafka实例）上。一般我们可以按照某个key的hash值去分partition,实现路由，具体的路由方式可以自行指定或者实现。然后，每个partition包含多个复本，分散在不同的broker,每个复本同步存储相同的log信息，保证高靠性。每个partition的复本组中有一个选作leader，而其他作follower,典型的行列式分布式布署。唯一让人觉着不痛快的，就是写和读都是走leader的，这样就无法把一些读负载均衡到follower上去。

并行与有序的矛盾

对于消息队列来说，并行与有序是矛盾的。假设，消息队列中存放的消息，是对数据库某表的内容修改操作命令，那么对同一条记录的修改操作命令必须有序到达，不然后面的结果选到，可能造成混乱，结果无意义。还是以active mq为例，满足这样的需求，要怎么办？没有办法，唯一的办法，就是保证一个queue，只有一个consumer在取，如果有多个consumer同时取的话，虽然consumer内部的消息能够保持有序，但是多个consumer之间的消息就无法保证有序了。这样的话，反正你只有一个consumer能取，再怎么分布式也是白搭，无法并行消费。
Kafka做了一定的改进。我们都知道，kafka的log存储是分partition的。而大多数有序需求，并不同要求全局有序。就像上文提到的要求，可能只要保证对同一个id的记录的操作保证有序便可。我们可以按照key(这里就是id值)，进行分组，将消息分到不同的partition中去，同一个id的相关纪录，肯定会归到同一个partition中去，而且在partition内部有序。这时就可以认为每一个partition就是一个单独的消息队列，可以为每个partition指定一个consumer。当然，如果为一个partition指定多个consumer又会丢失有序性。虽然不够完美，但相对传统jms，这种并行性的提高，已算是一个不小的进步。
那么如果你要求全局有序呢？抱歉，这种需求，kafka也只能通过指定一个单独的consumer来实现。幸好，一般的应用中很少出现这样的需求。按key分组，基本能满足大多数的需求。

终极一问：为什么kafka在consumer确认接收消息之后，还不删除消息，甚至提供consumer利用offset查找较早消息的功能？

我拿这个问题去问过我的几个不太熟悉kafka的逗比朋友，居然让他们折磨了一晚上也没想出来。我觉得为了理解kafka，必须要闹明白这个问题。
第一点，前文已述，kafka的存储方式，是按照数据文件（会按段划分）结合索引文件形成log来完成的，consumer用offset来查找，这种使用方式，注定不允许你对文件中的某条记录做删除操作。试想一下，你删了其中某条消息，你用来查的offset还会是对的吗？你是不是又要完全重新组织文件，想想就好烦。
第二点，就是确实存在consumer去找较老的消息的可能性存在。具体是什么场景呢？还是先上图吧

 这是一个最简单的生产者消费者模型。我们现在看到的消费者是一个完整的个体。消息队列，将消息发送给消费者，消费者反馈说已收到，消息队列就可以删消息了。确实很和谐，而且传统的jms就是这样做的。

但有的时候，消费者的处理并没有那么简单，消费者的处理可能分布式的处理，包含多个处理环节，第一个环节处理了，发送至下一个环节，下一个处理环节位于的可能就是不同的系统，已经是不同的服务器上了的进程了。当你第一个处理环节的节点接确认收到消息后，通知消息队列，已接收。那如果后续环节出现差错呢，比方如后面的传输中在到达终点前发现数据丢失，抑或是某个环节的服务挂掉了，这部份消息传输的可靠信又如何保证？难道你在每个处理节点之间再加具有能持久化功能，能保证消息可靠性的消息队列？这样想想，又是好复杂，好麻烦的样子。
利用kafka，就可以一直向第一个处理环节的节点发送消息，先不用管后续结点，当后续发现消息丢失的情况的下，就可以通过之前的offset，重新去从kafka获取这一条消息，全头重新执行（但是这样，存在有序性的问题）。刚才所述的多个处理环节的场景就是典型的流式计算的场景。这也是为什么storm流式计算框架官方推荐kafka作为其消息来源一个重要原因。
这部份属上个人理解，有要纠错的，或有补允的。欢迎在评论区留言。

最后，明天就是年三十了，祝各位读者老爷们，新年快乐！

20150217 首发于3dobe.com： http://3dobe.com/archives/68/

本站链接接： http://quentinxxz.iteye.com/blog/2186718