kakafka - 为CQRS而生

  前段时间跟一个朋友聊起kafka,flint,spark这些是否是某种分布式运算框架。我自认为的分布式运算框架最基础条件是可以把多个集群节点看成一个完整的系统，而后程序好像是在同一台机器的内存里运行同样。固然，这种集成实现方式有赖于底层的一套消息系统。这套消息系统能够把消息随意在集群各节点之间自由传递。因此若是可以经过消息来驱动某段程序的运行，那么这段程序就有可能在集群中任何一个节点上运行了。好了，akka-cluster是经过对每一个集群节点上的中介发送消息使之调动该节点上某段程序运行来实现分布式运算的。那么，kafka也能够实现消息在集群节点间的自由流通，是否是也是一个分布式运算框架呢？实际上，kafka设计强调的重点是消息的接收，或者叫消息消费机制。至于接收消息后怎么去应对，用什么方式处理，都是kafka用户本身的事了。与分布式运算框架像akka-cluster对比，kafka还缺了个在每一个集群节点上的”运算调度中介“，因此kafka应该不算我所指的分布式运算框架，充其量是一种分布式的消息传递系统。实际上kafka是一种高吞吐量、高可用性、安全稳定、有良好口碑的分布式消息系统。

  kafka的本质是一种commit-log，或者“事件记录系统”：上游产生的数据（即事件）会按发生时间顺序存入kafka，而后下游能够对任什么时候间段内事件按序进行读取，重演运算产生那段时间内的某种状态。这不就是妥妥的CQRS模式吗？固然kafka也可使用在其它一些场景如：消息队列，数据存储等，不过这些都是commit-log的具体应用。

  经常看到网上有朋友抱怨akka-cluster的一些处理方式太底层或太基础了。用户每每须要本身来增长一些方法来确保使用安全。我想做为一种消息驱动系统，如何保证akka消息的正确产生和安全使用应该是最基本的要求。而偏偏akka是没有提供对消息遗漏和重复消息的保障机制。我想这也是形成akka用户担忧的主要缘由。上面提到kafka是一种高吞吐量、高可用性、安全稳定的分布式消息系统，特别是它提供了对exactly-once,“保证一次”的消息使用支持。那么经过kafka实现一套CQRS模式的实时交易处理系统应该是可行的。这也是我使用kafka的主要目的。

  上面提到，但愿能充分利用kafka commit-log特性来开发一个基于CQRS的实时交易系统，好比支付系统、库存管理系统，从实践中了解kafka。kafka支持多种语言终端，怪异的是没有scala终端。kafka是用scala开发的，不提供scala终端实在是说不通啊。不过akka在alpakka社区提供了alpakka-kafka：这个东西是个基于akka-streams的kafka scala终端编程工具，稍微过了一下，感受功能比较全面，那就是它了。不过在开始前先把kafka的原理和基本状况作个介绍：

从表面上看kafka就是一个简单的消息存储和传递工具。不过由于其特殊分布式的消息发布、存储、读取处理机制，使其成为一种高吞吐量、高可用性、安全稳定的分布式消息处理工具。从应用角度来说，kafka应用包括三个方面，kafka自己，就叫kafka引擎吧，发布终端、订阅终端，即：kafka,writer,reader三部分，其中：全部复杂的功能实现是包嵌在kafka内部的，writer,reader应该整合到用户应用里。kafka的做业是围绕着消息的发布订阅/读写进行的。所谓消息即CQRS模式里的事件。那么kafka的工做原理直白点就是writer向kafka写事件，kafka把事件按发生时间顺序保存，reader再按顺序从kafka读取事件并进行处理以产生新的业务状态，如在某个库位的一个商品数量获得了更新。固然原理看似简单，但具体的实现才是真正复杂的地方。

首先，writer和reader是以事件关联的，即：write发布某种类型的事件，而reader则是订阅相同类型的事件。 这里的事件也就是topic，或一项业务，如：图书类当前库存。为了提升数据吞吐量，每一个topic又能够细分为多个partition。每一个partition分担所属topic消息类型下的一些指定的细分类消息或者事件，如"图书库房101"。若是把这些partition再分布到一个集群的节点上，就能够实现高吞吐量的分布式读写，而后经过集群partition的复本同步又能够达到数据安全及系统高可用性的目的。这些集群节点就是所谓的broker了。发布消息内容由topic,key,value所组成。其中key值指定该消息应该写入那个partition，即经过对key进行hash计算得出partition id。hash算法能够保证相同的key值永远指定同一个partition。值得注意的是kafka保证每一个partition上的事件确定按照发生时间排序，因此要保证一种事件只能写入同一个partition。固然，一个partition能够承载多种事件。要注意的是建立topic和partition都是严格的管理工做admin，不是在某些程序中任意进行增减的。通常来说，在建立一个新topic时就要肯定它下面的partition数量了。这个partition数量要按照对数据吞吐量需求设定。但通常是集群节点的倍数，这样partition能够均匀分布在各broker上。

好了，该到reader这头了：reader做业从订阅某个topic开始。上面提过：一个topic下面可能有多个partition，但每一个partition都会包含topic的其中几个子业务的所有事件，并且这些事件是严格按发生时间排序的。kafka有个reader group这么个概念：针对同一个topic，允许有一组多个reader对这个topic下的partition进行读取。但每一个partition只允许组内一个reader读取。至于goup内reader是如何分配partition的彻底由kafka内部解决。若是发现新partition或者组内reader有增减变化，kafka会自动进行再分配rebalance。因此总的来讲订阅某个topic的一个组内reader应该负责那个partition是不肯定的，加上随时可能发生动态再分配的状况，好比组内某个reader出问题倒了。换言之组内全部reader都必须具有处理整个topic全部类型业务的能力，如此才能解决组内reader-partition关系不肯定的难题。kafka最重要的特色就是能够允许不一样的应用经过不一样的reader-group对同一个partition上的事件进行任意读取，本意应该是不一样的应用能够利用同一个业务事件序列进行不一样的业务处理。具体实现方式应该是每一个组对某个partition上事件最后读取的位置分别进行了登记，offset-commit。这样，即便发生了从新分配rebalance组内任何一个reader对分配到的partition应从那个位置开始读仍是肯定的。这个offset-commit方式描述了几种事件读取模式：

一、at-most-once, 最多一次：若是刚读取事件，在进行业务处理以前就登记位置commit-offset，那么commit-offset后位置已经登记，即便业务处理失败也不再可能二次读取了。 

二、at-least-once，最少一次：读取事件、完成业务处理后才commit-offset。若是处理业务中系统故障，只能从上次登记的位置从新读取了，那么就会出现重复读取的状况。

三、exactly-once, 保证只一次：控制commit-offset的时间节点是取得at-most-once, at-least-once之间安全系数的一种方式。但exactly-once不允许有模糊地带。具体作法是把业务处理和commit-offset做为一个完整事物单元来处理（atomic-transaction)。两样操做同时成功或失败。

我觉着kafka的exactly-once能力最值得推介。由于在akka或者其它消息队列工具里不容易获得保证。而在一个消息驱动的实时交易系统里，保证事件重演能正确反映当时状态是关键。任何事件遗失或重复都会形成不可逆转的偏差。那么下面的一系列讨论我就会尝试用alpakka-kafka来构建一个基于CQRS模式的实时交易系统，并和你们进行交流分享。