kafka入门：简介、使用场景、设计原理、主要配置及集群搭建

问题导读：

1.zookeeper在kafka的作用是甚么？

2.kafka中几近不允许对消息进行“随机读写”的缘由是甚么？

3.kafka集群consumer和producer状态信息是如何保存的？

4.partitions设计的目的的根本缘由是甚么？

1、入门

    1、简介

    Kafka is a distributed,partitioned,replicated commit logservice。它提供了类似于JMS的特性，但是在设计实现上完全不同，另外它其实不是JMS规范的实现。kafka对消息保存时根据Topic进行归类，发送消息者成为Producer,消息接受者成为Consumer,另外kafka集群有多个kafka实例组成，每一个实例(server)成为broker。不管是kafka集群，还是producer和consumer都依赖于zookeeper来保证系统可用性集群保存1些meta信息。

 

   2、Topics/logs

    1个Topic可以认为是1类消息，每一个topic将被分成多个partition(区),每一个partition在存储层面是append log文件。任何发布到此partition的消息都会被直接追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），offset为1个long型数字，它是唯1标记1条消息。它唯1的标记1条消息。kafka并没有提供其他额外的索引机制来存储offset，由于在kafka中几近不允许对消息进行“随机读写”。

 

    kafka和JMS（Java Message Service）实现(activeMQ)不同的是:即便消息被消费,消息依然不会被立即删除.日志文件将会根据broker中的配置要求,保存1定的时间以后删除;比如log文件保存2天,那末两天后,文件会被清除,不管其中的消息是不是被消费.kafka通过这类简单的手段,来释放磁盘空间,和减少消息消费以后对文件内容改动的磁盘IO开支.

    对consumer而言,它需要保存消费消息的offset,对offset的保存和使用,有consumer来控制;当consumer正常消费消息时,offset将会"线性"的向先驱动,即消息将顺次顺序被消费.事实上consumer可使用任意顺序消费消息,它只需要将offset重置为任意值..(offset将会保存在zookeeper中,参见下文)

    kafka集群几近不需要保护任何consumer和producer状态信息,这些信息有zookeeper保存;因此producer和consumer的客户端实现非常轻量级,它们可以随便离开,而不会对集群造成额外的影响.

    partitions的设计目的有多个.最根本缘由是kafka基于文件存储.通过分区,可以将日志内容分散到多个server上,来避免文件尺寸到达单机磁盘的上限,每一个partiton都会被当前server(kafka实例)保存;可以将1个topic切分多任意多个partitions,来消息保存/消费的效力.另外越多的partitions意味着可以容纳更多的consumer,有效提升并发消费的能力.(具体原理参见下文).

    3、Distribution

    1个Topic的多个partitions,被散布在kafka集群中的多个server上;每一个server(kafka实例)负责partitions中消息的读写操作;另外kafka还可以配置partitions需要备份的个数(replicas),每一个partition将会被备份到多台机器上,以提高可用性.

    基于replicated方案,那末就意味着需要对多个备份进行调度;每一个partition都有1个server为"leader";leader负责所有的读写操作,如果leader失效,那末将会有其他follower来接收(成为新的leader);follower只是单调的和leader跟进,同步消息便可..因而可知作为leader的server承载了全部的要求压力,因此从集群的整体斟酌,有多少个partitions就意味着有多少个"leader",kafka会将"leader"均衡的分散在每一个实例上,来确保整体的性能稳定.

    Producers

    Producer将消息发布到指定的Topic中,同时Producer也能决定将此消息归属于哪一个partition;比如基于"round-robin"方式或通过其他的1些算法等.

    Consumers

    本质上kafka只支持Topic.每一个consumer属于1个consumer group;反过来讲,每一个group中可以有多个consumer.发送到Topic的消息,只会被定阅此Topic的每一个group中的1个consumer消费.

    如果所有的consumer都具有相同的group,这类情况和queue模式很像;消息将会在consumers之间负载均衡.

    如果所有的consumer都具有不同的group,那这就是"发布-定阅";消息将会广播给所有的消费者.

    在kafka中,1个partition中的消息只会被group中的1个consumer消费;每一个group中consumer消息消费相互独立;我们可以认为1个group是1个"定阅"者,1个Topic中的每一个partions,只会被1个"定阅者"中的1个consumer消费,不过1个consumer可以消费多个partitions中的消息.kafka只能保证1个partition中的消息被某个consumer消费时,消息是顺序的.事实上,从Topic角度来讲,消息仍不是有序的.

    kafka的设计原理决定,对1个topic,同1个group中不能有多于partitions个数的consumer同时消费,否则将意味着某些consumer将没法得到消息.

    Guarantees

    1) 发送到partitions中的消息将会依照它接收的顺序追加到日志中

    2) 对消费者而言,它们消费消息的顺序和日志中消息顺序1致.

    3) 如果Topic的"replicationfactor"为N,那末允许N⑴个kafka实例失效.

2、使用处景

    1、Messaging   

    对1些常规的消息系统,kafka是个不错的选择;partitons/replication和容错,可使kafka具有良好的扩大性和性能优势.不过到目前为止,我们应当很清楚认识到,kafka并没有提供JMS中的"事务性""消息传输担保(消息确认机制)""消息分组"等企业级特性;kafka只能使用作为"常规"的消息系统,在1定程度上,还没有确保消息的发送与接收绝对可靠(比如,消息重发,消息发送丢失等)

    2、Websit activity tracking

    kafka可以作为"网站活性跟踪"的最好工具;可以将网页/用户操作等信息发送到kafka中.并实时监控,或离线统计分析等

    3、Log Aggregation

    kafka的特性决定它非常合适作为"日志搜集中心";application可以将操作日志"批量""异步"的发送到kafka集群中,而不是保存在本地或DB中;kafka可以批量提交消息/紧缩消息等,这对producer端而言,几近感觉不到性能的开支.此时consumer端可使hadoop等其他系统化的存储和分析系统.

3、设计原理

    kafka的设计初衷是希望作为1个统1的信息搜集平台,能够实时的搜集反馈信息,并需要能够支持较大的数据量,且具有良好的容错能力.

    1、持久性

    kafka使用文件存储消息,这就直接决定kafka在性能上严重依赖文件系统的本身特性.且不管任何OS下,对文件系统本身的优化几近没有可能.文件缓存/直接内存映照等是经常使用的手段.由于kafka是对日志文件进行append操作,因此磁盘检索的开支是较小的;同时为了减少磁盘写入的次数,broker会将消息暂时buffer起来,当消息的个数(或尺寸)到达1定阀值时,再flush到磁盘,这样减少了磁盘IO调用的次数.

2、性能

    需要斟酌的影响性能点很多,除磁盘IO以外,我们还需要斟酌网络IO,这直接关系到kafka的吞吐量问题.kafka并没有提供太多高超的技能;对producer端,可以将消息buffer起来,当消息的条数到达1定阀值时,批量发送给broker;对consumer端也是1样,批量fetch多条消息.不过消息量的大小可以通过配置文件来指定.对kafka broker端,似乎有个sendfile系统调用可以潜伏的提升网络IO的性能:将文件的数据映照到系统内存中,socket直接读取相应的内存区域便可,而无需进程再次copy和交换. 其实对producer/consumer/broker3者而言,CPU的开支应当都不大,因此启用消息紧缩机制是1个良好的策略;紧缩需要消耗少许的CPU资源,不过对kafka而言,网络IO更应当需要斟酌.可以将任何在网络上传输的消息都经过紧缩.kafka支持gzip/snappy等多种紧缩方式.

    3、生产者

    负载均衡: producer将会和Topic下所有partition leader保持socket连接;消息由producer直接通过socket发送到broker,中间不会经过任何"路由层".事实上,消息被路由到哪一个partition上,有producer客户端决定.比如可以采取"random""key-hash""轮询"等,如果1个topic中有多个partitions,那末在producer端实现"消息均衡分发"是必要的.

    其中partition leader的位置(host:port)注册在zookeeper中,producer作为zookeeper client,已注册了watch用来监听partition leader的变更事件.

    异步发送：将多条消息暂且在客户端buffer起来，并将他们批量的发送到broker，小数据IO太多，会拖慢整体的网络延迟，批量延迟发送事实上提升了网络效力。不过这也有1定的隐患，比如说当producer失效时，那些还没有发送的消息将会丢失。

    4、消费者

    consumer端向broker发送"fetch"要求,并告知其获得消息的offset;尔后consumer将会取得1定条数的消息;consumer端也能够重置offset来重新消费消息.

    在JMS实现中,Topic模型基于push方式,即broker将消息推送给consumer端.不过在kafka中,采取了pull方式,即consumer在和broker建立连接以后,主动去pull(或说fetch)消息;这中模式有些优点,首先consumer端可以根据自己的消费能力适时的去fetch消息并处理,且可以控制消息消费的进度(offset);另外,消费者可以良好的控制消息消费的数量,batch fetch.

    其他JMS实现,消息消费的位置是有prodiver保存,以便避免重复发送消息或将没有消费成功的消息重发等,同时还要控制消息的状态.这就要求JMS broker需要太多额外的工作.在kafka中,partition中的消息只有1个consumer在消费,且不存在消息状态的控制,也没有复杂的消息确认机制,可见kafka broker端是相当轻量级的.当消息被consumer接收以后,consumer可以在本地保存最后消息的offset,并间歇性的向zookeeper注册offset.因而可知,consumer客户端也很轻量级.

 

    5、消息传送机制

    对JMS实现,消息传输担保非常直接:有且只有1次(exactly once).在kafka中稍有不同:

    1) at most once: 最多1次,这个和JMS中"非持久化"消息类似.发送1次,不管成败,将不会重发.

    2) at least once: 消息最少发送1次,如果消息未能接受成功,可能会重发,直到接收获功.

    3) exactly once: 消息只会发送1次.

    at most once: 消费者fetch消息,然后保存offset,然后处理消息;当client保存offset以后,但是在消息处理进程中出现了异常,致使部份消息未能继续处理.那末尔后"未处理"的消息将不能被fetch到,这就是"at most once".

    at least once: 消费者fetch消息,然后处理消息,然后保存offset.如果消息处理成功以后,但是在保存offset阶段zookeeper异常致使保存操作未能履行成功,这就致使接下来再次fetch时可能取得上次已处理过的消息,这就是"at least once"，缘由offset没有及时的提交给zookeeper，zookeeper恢复正常还是之前offset状态.

    exactly once: kafka中并没有严格的去实现(基于2阶段提交,事务),我们认为这类策略在kafka中是没有必要的.

    通常情况下"at-least-once"是我们搜选.(相比at most once而言,重复接收数据总比丢失数据要好).

    6、复制备份

    kafka将每一个partition数据复制到多个server上,任何1个partition有1个leader和多个follower(可以没有);备份的个数可以通过broker配置文件来设定.leader处理所有的read-write要求,follower需要和leader保持同步.Follower和consumer1样,消费消息并保存在本地日志中;leader负责跟踪所有的follower状态,如果follower"落后"太多或失效,leader将会把它从replicas同步列表中删除.当所有的follower都将1条消息保存成功,此消息才被认为是"committed",那末此时consumer才能消费它.即便只有1个replicas实例存活,依然可以保证消息的正常发送和接收,只要zookeeper集群存活便可.(不同于其他散布式存储,比如hbase需要"多数派"存活才行)

    当leader失效时,需在followers当选取出新的leader,可能此时follower落后于leader,因此需要选择1个"up-to-date"的follower.选择follower时需要统筹1个问题,就是新leaderserver上所已承载的partition leader的个数,如果1个server上有过量的partition leader,意味着此server将承受着更多的IO压力.在选举新leader,需要斟酌到"负载均衡".

    7.日志

    如果1个topic的名称为"my_topic",它有2个partitions,那末日志将会保存在my_topic_0和my_topic_1两个目录中;日志文件中保存了1序列"log entries"(日志条目),每一个log entry格式为"4个字节的数字N表示消息的长度" + "N个字节的消息内容";每一个日志都有1个offset来唯1的标记1条消息,offset的值为8个字节的数字,表示此消息在此partition中所处的起始位置..每一个partition在物理存储层面,有多个log file组成(称为segment).segmentfile的命名为"最小offset".kafka.例如"00000000000.kafka";其中"最小offset"表示此segment中起始消息的offset.

 

    其中每一个partiton中所持有的segments列表信息会存储在zookeeper中.

    当segment文件尺寸到达1定阀值时(可以通过配置文件设定,默许1G),将会创建1个新的文件;当buffer中消息的条数到达阀值时将会触发日志信息flush到日志文件中,同时如果"距离最近1次flush的时间差"到达阀值时,也会触发flush到日志文件.如果broker失效,极有可能会丢失那些还没有flush到文件的消息.由于server意外实现,依然会致使log文件格式的破坏(文件尾部),那末就要求当server启东是需要检测最后1个segment的文件结构是不是合法并进行必要的修复.

    获得消息时,需要指定offset和最大chunk尺寸,offset用来表示消息的起始位置,chunk size用来表示最大获得消息的总长度(间接的表示消息的条数).根据offset,可以找到此消息所在segment文件,然后根据segment的最小offset取差值,得到它在file中的相对位置,直接读取输出便可.

    日志文件的删除策略非常简单:启动1个后台线程定期扫描log file列表,把保存时间超过阀值的文件直接删除(根据文件的创建时间).为了不删除文件时依然有read操作(consumer消费),采取copy-on-write方式.

    8、分配

    kafka使用zookeeper来存储1些meta信息,并使用了zookeeper watch机制来发现meta信息的变更并作出相应的动作(比如consumer失效,触发负载均衡等)

    1) Broker node registry: 当1个kafkabroker启动后,首先会向zookeeper注册自己的节点信息(临时znode),同时当broker和zookeeper断开连接时,此znode也会被删除.

    格式: /broker/ids/[0...N]   -->host:port;其中[0..N]表示broker id,每一个broker的配置文件中都需要指定1个数字类型的id(全局不可重复),znode的值为此broker的host:port信息.

    2) Broker Topic Registry: 当1个broker启动时,会向zookeeper注册自己持有的topic和partitions信息,依然是1个临时znode.

    格式: /broker/topics/[topic]/[0...N]  其中[0..N]表示partition索引号.

    3) Consumer and Consumer group: 每一个consumer客户端被创建时,会向zookeeper注册自己的信息;此作用主要是为了"负载均衡".

    1个group中的多个consumer可以交错的消费1个topic的所有partitions;简而言之,保证此topic的所有partitions都能被此group所消费,且消费时为了性能斟酌,让partition相对均衡的分散到每一个consumer上.

    4) Consumer id Registry: 每一个consumer都有1个唯1的ID(host:uuid,可以通过配置文件指定,也能够由系统生成),此id用来标记消费者信息.

    格式:/consumers/[group_id]/ids/[consumer_id]

    依然是1个临时的znode,此节点的值为{"topic_name":#streams...},即表示此consumer目前所消费的topic + partitions列表.

    5) Consumer offset Tracking: 用来跟踪每一个consumer目前所消费的partition中最大的offset.

    格式:/consumers/[group_id]/offsets/[topic]/[broker_id-partition_id]-->offset_value

    此znode为持久节点,可以看出offset跟group_id有关,以表明当group中1个消费者失效,其他consumer可以继续消费.

    6) Partition Owner registry: 用来标记partition被哪一个consumer消费.临时znode

    格式:/consumers/[group_id]/owners/[topic]/[broker_id-partition_id]-->consumer_node_id当consumer启动时,所触发的操作:

    A) 首先进行"Consumer id Registry";

    B) 然后在"Consumer id Registry"节点下注册1个watch用来监听当前group中其他consumer的"leave"和"join";只要此znode path下节点列表变更,都会触发此group下consumer的负载均衡.(比如1个consumer失效,那末其他consumer接收partitions).

    C) 在"Broker id registry"节点下,注册1个watch用来监听broker的存活情况;如果broker列表变更,将会触发所有的groups下的consumer重新balance.

 

    1) Producer端使用zookeeper用来"发现"broker列表,和和Topic下每一个partition leader建立socket连接并发送消息.

    2) Broker端使用zookeeper用来注册broker信息,已监测partitionleader存活性.

    3) Consumer端使用zookeeper用来注册consumer信息,其中包括consumer消费的partition列表等,同时也用来发现broker列表,并和partition leader建立socket连接,并获得消息.

4、主要配置

    1、Broker配置

 

    2.Consumer主要配置

 

3.Producer主要配置

 

以上是关于kafka1些基础说明，在其中我们知道如果要kafka正常运行，必须配置zookeeper，否则不管是kafka集群还是客户真个生存者和消费者都没法正常的工作的，以下是对zookeeper进行1些简单的介绍：

5、zookeeper集群

    zookeeper是1个为散布式利用提供1致性服务的软件，它是开源的Hadoop项目的1个子项目，并根据google发表的1篇论文来实现的。zookeeper为散布式系统提供了高笑且易于使用的协同服务，它可以为散布式利用提供相当多的服务，诸如统1命名服务，配置管理，状态同步和组服务等。zookeeper接口简单，我们没必要过量地纠结在散布式系统编程难于处理的同步和1致性问题上，你可使用zookeeper提供的现成(off-the-shelf)服务来实现来实现散布式系统额配置管理，组管理，Leader选举等功能。

    zookeeper集群的安装,准备3台服务器server1:192.168.0.1,server2:192.168.0.2,

    server3:192.168.0.3.

    1)下载zookeeper

    到http://zookeeper.apache.org/releases.html去下载最新版本Zookeeper⑶.4.5的安装包zookeeper⑶.4.5.tar.gz.将文件保存server1的~目录下

    2)安装zookeeper

    先在服务器server分别履行a-c步骤

    a)解压  

    tar -zxvf zookeeper⑶.4.5.tar.gz

    解压完成后在目录~下会发现多出1个目录zookeeper⑶.4.5,重新命令为zookeeper

    b）配置

    将conf/zoo_sample.cfg拷贝1份命名为zoo.cfg，也放在conf目录下。然后依照以下值修改其中的配置：

   

    # The number of milliseconds of each tick

    tickTime=2000

    # The number of ticks that the initial

    # synchronization phase can take

    initLimit=10

    # The number of ticks that can pass between

    # sending a request and getting an acknowledgement

    syncLimit=5

    # the directory where the snapshot is stored.

    # do not use /tmp for storage, /tmp here is just

    # example sakes.

    dataDir=/home/wwb/zookeeper /data

    dataLogDir=/home/wwb/zookeeper/logs

    # the port at which the clients will connect

    clientPort=2181

    #

    # Be sure to read the maintenance section of the

    # administrator guide before turning on autopurge.

    #http://zookeeper.apache.org/doc/ ... html#sc_maintenance

    #

    # The number of snapshots to retain in dataDir

    #autopurge.snapRetainCount=3

    # Purge task interval in hours

    # Set to "0" to disable auto purge feature

    #autopurge.purgeInterval=1

    server.1=192.168.0.1:3888:4888

    server.2=192.168.0.2:3888:4888

    server.3=192.168.0.3:3888:4888

    tickTime：这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间保持心跳的时间间隔，也就是每一个 tickTime 时间就会发送1个心跳。

    dataDir：顾名思义就是 Zookeeper 保存数据的目录，默许情况下，Zookeeper 将写数据的日志文件也保存在这个目录里。

    clientPort：这个端口就是客户端连接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户真个访问要求。

    initLimit：这个配置项是用来配置 Zookeeper 接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器集群中连接到 Leader 的 Follower 服务器）初始化连接时最长能忍耐多少个心跳时间间隔数。当已超过 5个心跳的时间（也就是 tickTime）长度后 Zookeeper 服务器还没有收到客户真个返回信息，那末表明这个客户端连接失败。总的时间长度就是 5*2000=10 秒

    syncLimit：这个配置项标识 Leader 与Follower 之间发送消息，要求和应对时间长度，最长不能超过量少个 tickTime 的时间长度，总的时间长度就是2*2000=4 秒

    server.A=B：C：D：其中 A 是1个数字，表示这个是第几号服务器；B 是这个服务器的 ip 地址；C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；D 表示的是万1集群中的 Leader 服务器挂了，需要1个端口来重新进行选举，选出1个新的 Leader，而这个端口就是用来履行选举时服务器相互通讯的端口。如果是伪集群的配置方式，由于 B 都是1样，所以不同的 Zookeeper 实例通讯端口号不能1样，所以要给它们分配不同的端口号

注意:dataDir,dataLogDir中的wwb是当前登录用户名，data，logs目录开始是不存在，需要使用mkdir命令创建相应的目录。并且在该目录下创建文件myid,serve1,server2,server3该文件内容分别为1,2,3。

针对服务器server2,server3可以将server1复制到相应的目录，不过需要注意dataDir,dataLogDir目录,并且文件myid内容分别为2,3

    3)顺次启动server1，server2,server3的zookeeper.

    /home/wwb/zookeeper/bin/zkServer.sh start,出现类似以下内容

    JMX enabled by default

    Using config: /home/wwb/zookeeper/bin/../conf/zoo.cfg

    Starting zookeeper ... STARTED

   4) 测试zookeeper是不是正常工作，在server1上履行以下命令

    /home/wwb/zookeeper/bin/zkCli.sh -server192.168.0.2:2181,出现类似以下内容

    JLine support is enabled

    2013⑴1⑵7 19:59:40,560 - INFO      [main-SendThread(localhost.localdomain:2181):ClientCnxn$SendThread@736]- Session   establishmentcomplete on server localhost.localdomain/127.0.0.1:2181, sessionid =    0x1429cdb49220000, negotiatedtimeout = 30000

    WATCHER::

   

    WatchedEvent state:SyncConnected type:None path:null

    [zk: 127.0.0.1:2181(CONNECTED) 0] [root@localhostzookeeper2]#  

    即代表集群构建成功了,如果出现毛病那应当是第3部时没有启动好集群，

运行，先利用

    ps aux | grep zookeeper查看是不是有相应的进程的，没有话，说明集群启动出现问题，可以在每一个服务器上使用

    ./home/wwb/zookeeper/bin/zkServer.sh stop。再顺次使用./home/wwb/zookeeper/binzkServer.sh start，这时候在履行41般是没有问题，如果还是有问题，那末先stop再到bin的上级目录履行./bin/zkServer.shstart试试。

注意：zookeeper集群时，zookeeper要求半数以上的机器可用，zookeeper才能提供服务。