1. 软件开发中的Kafka和数据库的关系是什么呢
首先明确说明Kafka不是数据库，它没有schema，也没有表，更没有索引。
1.它仅仅是生产消息流、消费消息流而已。从这个角度来说Kafka的确不像数据库，至少不像我们熟知的关系型数据库。
那么到底什么是数据库呢？或者说什么特性使得一个系统可以被称为数据库？经典的教科书是这么说的：数据库是提供 ACID 特性的，我们依次讨论下ACID。
1、持久性(rability)
我们先从最容易的持久性开始说起，因为持久性最容易理解。在80年代持久性指的是把数据写入到磁带中，这是一种很古老的存储设备，现在应该已经绝迹了。目前实现持久性更常见的做法是将数据写入到物理磁盘上，而这也只能实现单机的持久性。当演进到分布式系统时代后，持久性指的是将数据通过备份机制拷贝到多台机器的磁盘上。很多数据库厂商都有自己的分布式系统解决方案，如GreenPlum和Oracle RAC。它们都提供了这种多机备份的持久性。和它们类似，Apache Kafka天然也是支持这种持久性的，它提供的副本机制在实现原理上几乎和数据库厂商的方案是一样的。
2、原子性(atomicity)
数据库中的原子性和多线程领域内的原子性不是一回事。我们知道在Java中有AtomicInteger这样的类能够提供线程安全的整数操作服务，这里的atomicity关心的是在多个线程并发的情况下如何保证正确性的问题。而在数据库领域，原子性关心的是如何应对错误或异常情况，特别是对于事务的处理。如果服务发生故障，之前提交的事务要保证已经持久化，而当前运行的事务要终止(abort)，它执行的所有操作都要回滚，最终的状态就好像该事务从未运行过那样。举个实际的例子，
第三个方法是采用基于日志结构的消息队列来实现，比如使用Kafka来做，如下图所示：
在这个架构中app仅仅是向Kafka写入消息，而下面的数据库、cache和index作为独立的consumer消费这个日志——Kafka分区的顺序性保证了app端更新操作的顺序性。如果某个consumer消费速度慢于其他consumer也没关系，毕竟消息依然在Kafka中保存着。总而言之，有了Kafka所有的异质系统都能以相同的顺序应用app端的更新操作，
3、隔离性(isolation)
在传统的关系型数据库中最强的隔离级别通常是指serializability，国内一般翻译成可串行化或串行化。表达的思想就是连接数据库的每个客户端在执行各自的事务时数据库会给它们一个假象：仿佛每个客户端的事务都顺序执行的，即执行完一个事务之后再开始执行下一个事务。其实数据库端同时会处理多个事务，但serializability保证了它们就像单独执行一样。举个例子，在一个论坛系统中，每个新用户都需要注册一个唯一的用户名。一个简单的app实现逻辑大概是这样的：
4、一致性(consistency)
最后说说一致性。按照Kelppmann大神的原话，这是一个很奇怪的属性：在所有ACID特性中，其他三项特性的确属于数据库层面需要实现或保证的，但只有一致性是由用户来保证的。严格来说，它不属于数据库的特性，而应该属于使用数据库的一种方式。坦率说第一次听到这句话时我本人还是有点震惊的，因为从没有往这个方面考虑过，但仔细想想还真是这么回事。比如刚才的注册用户名的例子中我们要求每个用户名是唯一的。这种一致性约束是由我们用户做出的，而不是数据库本身。数据库本身并不关心或并不知道用户名是否应该是唯一的。针对Kafka而言，这种一致性又意味着什么呢？Kelppmann没有具体展开，
希望能帮到你，谢谢！
2. “根本就不需要 Kafka 这样的大型分布式系统！”
作者 | Normcore Tech
译者 | 弯月，责编 | 屠敏
出品 | CSDN（ID：CSDNnews）
以下为译文：
可能有人没有听说过Kafka，这是一个非常复杂的分布式软件，可协调多台计算机之间的数据传输。更具体地说，该软件的功能是“展平”数据，然后快速地将数据从一个地方移动到另一个地方。一般来讲，如果你有很多数据需要快速处理并发送到其他地方，那么就可以考虑一下Kafka。Kafka还可以在一定期限内保留数据，比如设置数据保存2天、3天或7天，如果你的下游流程失败，那么你还可以利用存储在Kafka中的数据重新处理。
许多处理汇总数据的公司（比如Facebook和Twitter等社交网络数据，以及每晚需要处理大量星体运动的天文学家，或需要快速了解车辆周围环境数据的自动驾驶车辆公司等）都在使用Kafka，将任意地方生产的数据（即用户通过键盘输入的数据，通过望远镜读取的数据，通过车辆遥测读取的数据等）移动至下游流程进行处理和分析。
最近，WeWork更为名The We Company，他们在共享工作间领域取得了成功，其官网宣称公司的使命为：
“提升世界的意识。”其核心业务是从房地产出租公司那里租下办公室，然后转租给无法按照传统流程租赁办公室的个人和小公司。
为了“提升世界的意识”，该公司致力于为世界各地的个人和公司的团队打造独特却又不完全相同的办公空间。最近，该公司还开始涉足教育。
最近，因为上市，WeWork曝光了一些财务信息：
从好的方面来看，根据A xi os的数据，2018年WeWork的入住率为90%，且会员总数在不断增加。
有人常常将WeWork视为硅谷地区的公司过高估值的完美例子。作为一家房地产企业，WeWork烧钱的速度非常快，毫无疑问他们必须努力让公众市场投资者相信公司有长远的发展，同时和还要维护其作为 科技 公司的地位。
这家公司再三强调说它不是一家房地产公司（毕竟它在不断烧钱对吧？），那么一家消息中介技术公司究竟能提供什么？WeWork曾宣布，它使用Kafka来实现“内部部署的物联网需求”。这是什么意思？
“我们的产品是物理空间，”WeWork的首席开发负责人David Fano说，他在会议期间穿着一件印有“bldgs = data”字样的T恤。
每个办公室都有10个环境传感器——小巧的壁挂式绿色盒子，这些传感器可跟踪室内温度、湿度、空气质量、气压和环境光线水平。还有20个白色的壁挂式信标，呈三角形分布在公共空间（开放式办公区和会议室），用于测量WeWork成员的室内位置（数据是匿名的）。顶部四分之一的传感器通过计算机视觉观察成员的活动。
换句话说，WeWork会跟踪WeWork的多个物理事件并记录所有这些数据。但是......他们真的有必要这样做吗？记录Keith Harring壁画周围开放区域的环境温度能给他们带来怎样的竞争优势？更重要的是，他们能否将这些信息用到重要的项目中？
对于公司而言，重要的是要了解办公室的“单位组合” ——私人办公室、会议空间和开放式办公桌——的比例，我们可以利用这些信息对下一个办公间作出调整。
我觉得这家新闻报道机构需要建立一种思考技术的心理模型。Ben Thompson为Stratechery提供了出色的服务，他建立了聚合理论（https://stratechery .com /concepts/），我在努力为这些理论建立一个网站，如果必须从中选择一个的话，那便是：
大多数创业公司（以及大公司）现有的技术栈都没有必要。
在此，我想挑战一下那些自认为可以在一个周末期间独自建立Facebook的Hacker News上的开发人员，我认为WeWork的实际业务和架构问题在于：
WeWork需要的只不过是清点进出的人数，然后对容量规划做优化而已，追踪“气压”有什么用？只要你有WeWork的ID，那你肯定是个人或公司。那么，在大堂里安装一个登记系统，并要求会议系统发放名牌，不是更简单吗？
第一项要求根本就不需要Kafka：
目前WeWork有280个办公间。假设每个办公间平均每天有1000个（有这么多吗？）成员出入。那么每天会产生280,000个事务。我们假设每个人在早餐时间进来一次，在午餐时间出入各一次，然后离开。那么每个人会产生4个事务。那么每天大约是100万个事务，这点数据量存储在最常用的开源关系数据库Postgres中就可以了。保守地说，Postgres每秒可以提供10,000次写入（如果设置得当，其写入次数会更高）。每天100万个事件，也就是每秒11次。根本就不是问题。
至于第二项要求，受预订会议室人数的影响，产生的数据量可能更高，但你不需要实时传输数据。你完全可以等到一天结束时批量处理或收集，这同样可以利用司空见惯的关系数据库。
与大型Postgres（或者是BigQuery，或选择其他关系数据库连接到接收JSON传感器数据的Web服务）相比，Kafka的日常开销要高出很多，因为分布式系统非常非常复杂，比传统的系统复杂得多。
Kafka是一个非常优秀的强大的工具，但各个公司在采用该软件时，需要三思而后行。杀鸡焉用牛刀，WeWork用Kafka来记录开放办公间的气压，实属大材小用。
虽然很多时候我们都不需要Kafka，但开发人员很喜欢推荐这个工具，因为他们可以借机积攒经验和谈资。开发人员喜欢用最尖端的技术来完成工作，有时甚至他们自己都没意识到这一点。
过度架构真实存在。 Nemil在一篇文章中说：
在职业生涯的早期，你遇到的大量设计不良的软件系统都要归咎于那些传播错误观点的工程媒体。
在大学和培训班中，你对工程的了解主要来自工程媒体，例如 Hacker News、聚会、会议、Free Code Camp和Hacker Noon等。这些网站广泛讨论的技术（比如微服务、前端框架或区块链）自然会现在你的技术栈中，虽然不是很必要。
使用这些技术栈会导致各个公司承担不必要的债务，导致他们不得不在风险投资周期中寻求更多的资金，无法迈向精益或从别人的资金中解脱出来。
这种不幸的趋势只会持续下去，我们唯一能做的就是公之于众。
原文：https://vicki.substack .com /p/you-dont-need-kafka
【END】
3. Kafka核心组件之控制器和协调器
[TOC]
我们已经知道Kafka的集群由n个的broker所组成，每个broker就是一个kafka的实例或者称之为kafka的服务。其实控制器也是一个broker，控制器也叫leader broker。
他除了具有一般broker的功能外，还负责分区leader的选取，也就是负责选举partition的leader replica。
kafka每个broker启动的时候，都会实例化一个KafkaController，并将broker的id注册到zookeeper，集群在启动过程中，通过选举机制选举出其中一个broker作为leader，也就是前面所说的控制器。
包括集群启动在内，有三种情况触发控制器选举：
1、集群启动
2、控制器所在代理发生故障
3、zookeeper心跳感知，控制器与自己的session过期
按照惯例，先看图。我们根据下图来讲解集群启动时，控制器选举过程。
假设此集群有三个broker，同时启动。
（一）3个broker从zookeeper获取/controller临时节点信息。/controller存储的是选举出来的leader信息。此举是为了确认是否已经存在leader。
（二）如果还没有选举出leader，那么此节点是不存在的，返回-1。如果返回的不是-1，而是leader的json数据，那么说明已经有leader存在，选举结束。
（三）三个broker发现返回-1，了解到目前没有leader，于是均会触发向临时节点/controller写入自己的信息。最先写入的就会成为leader。
（四）假设broker 0的速度最快，他先写入了/controller节点，那么他就成为了leader。而broker1、broker2很不幸，因为晚了一步，他们在写/controller的过程中会抛出ZkNodeExistsException，也就是zk告诉他们，此节点已经存在了。
经过以上四步，broker 0成功写入/controller节点，其它broker写入失败了，所以broker 0成功当选leader。
此外zk中还有controller_epoch节点，存储了leader的变更次数，初始值为0，以后leader每变一次，该值+1。所有向控制器发起的请求，都会携带此值。如果控制器和自己内存中比较，请求值小，说明kafka集群已经发生了新的选举，此请求过期，此请求无效。如果请求值大于控制器内存的值，说明已经有新的控制器当选了，自己已经退位，请求无效。kafka通过controller_epoch保证集群控制器的唯一性及操作的一致性。
由此可见，Kafka控制器选举就是看谁先争抢到/controller节点写入自身信息。
控制器的初始化，其实是初始化控制器所用到的组件及监听器，准备元数据。
前面提到过每个broker都会实例化并启动一个KafkaController。KafkaController和他的组件关系，以及各个组件的介绍如下图：
图中箭头为组件层级关系，组件下面还会再初始化其他组件。可见控制器内部还是有些复杂的，主要有以下组件：
1、ControllerContext，此对象存储了控制器工作需要的所有上下文信息，包括存活的代理、所有主题及分区分配方案、每个分区的AR、leader、ISR等信息。
2、一系列的listener，通过对zookeeper的监听，触发相应的操作，黄色的框的均为listener
3、分区和副本状态机，管理分区和副本。
4、当前代理选举器ZookeeperLeaderElector，此选举器有上位和退位的相关回调方法。
5、分区leader选举器，PartitionLeaderSelector
6、主题删除管理器，TopicDeletetionManager
7、leader向broker批量通信的ControllerBrokerRequestBatch。缓存状态机处理后产生的request，然后统一发送出去。
8、控制器平衡操作的KafkaScheler，仅在broker作为leader时有效。
Kafka集群的一些重要信息都记录在ZK中，比如集群的所有代理节点、主题的所有分区、分区的副本信息(副本集、主副本、同步的副本集)。每个broker都有一个控制器，为了管理整个集群Kafka选利用zk选举模式，为整个集群选举一个“中央控制器”或”主控制器“，控制器其实就是一个broker节点，除了一般broker功能外，还具有分区首领选举功能。中央控制器管理所有节点的信息，并通过向ZK注册各种监听事件来管理整个集群节点、分区的leader的选举、再平衡等问题。外部事件会更新ZK的数据，ZK中的数据一旦发生变化，控制器都要做不同的响应处理。
故障转移其实就是leader所在broker发生故障，leader转移为其他的broker。转移的过程就是重新选举leader的过程。
重新选举leader后，需要为该broker注册相应权限，调用的是ZookeeperLeaderElector的onControllerFailover()方法。在这个方法中初始化和启动了一系列的组件来完成leader的各种操作。具体如下，其实和控制器初始化有很大的相似度。
1、注册分区管理的相关监听器
2、注册主题管理的相关监听
3、注册代理变化监听器
4、重新初始化ControllerContext，
5、启动控制器和其他代理之间通信的ControllerChannelManager
6、创建用于删除主题的TopicDeletionManager对象,并启动。
7、启动分区状态机和副本状态机
8、轮询每个主题，添加监听分区变化的
9、如果设置了分区平衡定时操作，那么创建分区平衡的定时任务，默认300秒检查并执行。
除了这些组件的启动外，onControllerFailover方法中还做了如下操作：
1、/controller_epoch值+1，并且更新到ControllerContext
2、检查是否出发分区重分配，并做相关操作
3、检查需要将优先副本选为leader，并做相关操作
4、向kafka集群所有代理发送更新元数据的请求。
下面来看leader权限被取消时，调用的方法onControllerResignation
1、该方法中注销了控制器的权限。取消在zookeeper中对于分区、副本感知的相应监听器的监听。
2、关闭启动的各个组件
3、最后把ControllerContext中记录控制器版本的数值清零，并设置当前broker为RunnignAsBroker，变为普通的broker。
通过对控制器启动过程的学习，我们应该已经对kafka工作的原理有了了解， 核心是监听zookeeper的相关节点，节点变化时触发相应的操作 。
有新的broker加入集群时，称为代理上线。反之，当broker关闭，推出集群时，称为代理下线。
代理上线：
1、新代理启动时向/brokers/ids写数据
2、BrokerChangeListener监听到变化。对新上线节点调用controllerChannelManager.addBroker()，完成新上线代理网络层初始化
3、调用KafkaController.onBrokerStartup()处理
3.5恢复因新代理上线暂停的删除主题操作线程
代理下线：
1、查找下线节点集合
2、轮询下线节点，调用controllerChannelManager.removeBroker()，关闭每个下线节点网络连接。清空下线节点消息队列，关闭下线节点request请求
3、轮询下线节点，调用KafkaController.onBrokerFailure处理
4、向集群全部存活代理发送updateMetadataRequest请求
顾名思义，协调器负责协调工作。本节所讲的协调器，是用来协调消费者工作分配的。简单点说，就是消费者启动后，到可以正常消费前，这个阶段的初始化工作。消费者能够正常运转起来，全有赖于协调器。
主要的协调器有如下两个：
1、消费者协调器（ConsumerCoordinator）
2、组协调器（GroupCoordinator）
kafka引入协调器有其历史过程，原来consumer信息依赖于zookeeper存储，当代理或消费者发生变化时，引发消费者平衡，此时消费者之间是互不透明的，每个消费者和zookeeper单独通信，容易造成羊群效应和脑裂问题。
为了解决这些问题，kafka引入了协调器。服务端引入组协调器（GroupCoordinator），消费者端引入消费者协调器（ConsumerCoordinator）。每个broker启动的时候，都会创建GroupCoordinator实例，管理部分消费组（集群负载均衡）和组下每个消费者消费的偏移量（offset）。每个consumer实例化时，同时实例化一个ConsumerCoordinator对象，负责同一个消费组下各个消费者和服务端组协调器之前的通信。如下图：
消费者协调器，可以看作是消费者做操作的代理类（其实并不是），消费者很多操作通过消费者协调器进行处理。
消费者协调器主要负责如下工作：
1、更新消费者缓存的MetaData
2、向组协调器申请加入组
3、消费者加入组后的相应处理
4、请求离开消费组
5、向组协调器提交偏移量
6、通过心跳，保持组协调器的连接感知。
7、被组协调器选为leader的消费者的协调器，负责消费者分区分配。分配结果发送给组协调器。
8、非leader的消费者，通过消费者协调器和组协调器同步分配结果。
消费者协调器主要依赖的组件和说明见下图：
可以看到这些组件和消费者协调器担负的工作是可以对照上的。
组协调器负责处理消费者协调器发过来的各种请求。它主要提供如下功能：
组协调器在broker启动的时候实例化，每个组协调器负责一部分消费组的管理。它主要依赖的组件见下图：
这些组件也是和组协调器的功能能够对应上的。具体内容不在详述。
下图展示了消费者启动选取leader、入组的过程。
消费者入组的过程，很好的展示了消费者协调器和组协调器之间是如何配合工作的。leader consumer会承担分区分配的工作，这样kafka集群的压力会小很多。同组的consumer通过组协调器保持同步。消费者和分区的对应关系持久化在kafka内部主题。
消费者消费时，会在本地维护消费到的位置（offset），就是偏移量，这样下次消费才知道从哪里开始消费。如果整个环境没有变化，这样做就足够了。但一旦消费者平衡操作或者分区变化后，消费者不再对应原来的分区，而每个消费者的offset也没有同步到服务器，这样就无法接着前任的工作继续进行了。
因此只有把消费偏移量定期发送到服务器，由GroupCoordinator集中式管理，分区重分配后，各个消费者从GroupCoordinator读取自己对应分区的offset，在新的分区上继续前任的工作。
下图展示了不提交offset到服务端的问题：
开始时，consumer 0消费partition 0 和1，后来由于新的consumer 2入组，分区重新进行了分配。consumer 0不再消费partition2，而由consumer 2来消费partition 2，但由于consumer之间是不能通讯的，所有consumer2并不知道从哪里开始自己的消费。
因此consumer需要定期提交自己消费的offset到服务端，这样在重分区操作后，每个consumer都能在服务端查到分配给自己的partition所消费到的offset，继续消费。
由于kafka有高可用和横向扩展的特性，当有新的分区出现或者新的消费入组后，需要重新分配消费者对应的分区，所以如果偏移量提交的有问题，会重复消费或者丢消息。偏移量提交的时机和方式要格外注意！！
1、自动提交偏移量
设置 enable.auto.commit为true，设定好周期，默认5s。消费者每次调用轮询消息的poll() 方法时，会检查是否超过了5s没有提交偏移量，如果是，提交上一次轮询返回的偏移量。
这样做很方便，但是会带来重复消费的问题。假如最近一次偏移量提交3s后，触发了再均衡，服务器端存储的还是上次提交的偏移量，那么再均衡结束后，新的消费者会从最后一次提交的偏移量开始拉取消息，此3s内消费的消息会被重复消费。
2、手动提交偏移量
设置 enable.auto.commit为false。程序中手动调用commitSync()提交偏移量，此时提交的是poll方法返回的最新的偏移量。
commitSync()是同步提交偏移量，主程序会一直阻塞，偏移量提交成功后才往下运行。这样会限制程序的吞吐量。如果降低提交频次，又很容易发生重复消费。
这里我们可以使用commitAsync()异步提交偏移量。只管提交，而不会等待broker返回提交结果
commitSync只要没有发生不可恢复错误，会进行重试，直到成功。而commitAsync不会进行重试，失败就是失败了。commitAsync不重试，是因为重试提交时，可能已经有其它更大偏移量已经提交成功了，如果此时重试提交成功，那么更小的偏移量会覆盖大的偏移量。那么如果此时发生再均衡，新的消费者将会重复消费消息。