kafka

一、Kafka简介

Kafka是一个分布式的消息队列系统(Message Queue)。

官网：https://kafka.apache.org/

 

kafka集群有多个Broker(不揉克)服务器组成，每个类型的消息被定义为topic（套屁）。

同一topic内部的消息按照一定的key和算法被分区(partition)存储在不同的Broker上。

消息生产者producer和消费者consumer可以在多个Broker上生产/消费topic

 

概念理解：

Ø  Topics and Logs：

Topic即为每条发布到Kafka集群的消息都有一个类别，topic在Kafka中可以由多个消费者订阅、消费。每个topic包含一个或多个partition（分区），partition数量可以在创建topic时指定，每个分区日志中记录了该分区的数据以及索引信息。如下图：

 

Kafka只保证一个分区内的消息有序，不能保证一个主题的不同分区之间的消息有序。如果你想要保证所有的消息都绝对有序可以只为一个主题分配一个分区。

分区会给每个消息记录分配一个顺序ID号（偏移量），能够唯一地标识该分区中的每个记录。Kafka集群保留所有发布的记录，不管这个记录有没有被消费过，Kafka提供相应策略通过配置从而对旧数据处理。

 

实际上，每个消费者唯一保存的元数据信息就是消费者当前消费日志的位移位置。位移位置是由消费者控制，即、消费者可以通过修改偏移量读取任何位置的数据。

 

Ø  Distribution -- 分布式

Ø  Producers -- 生产者，指定topic来发送消息到Kafka Broker

Ø  Consumers -- 消费者，根据topic消费相应的消息

 

第二：kafka的基础知识

1.       Producer生产者自己决定往哪个partition写消息，可以是轮询的负载均衡，或者是基于hash的（hash取模的做法容易产生数据倾斜）。一个topic可以分成多个partition，在创建topic的时候可以指定partition的个数。

2.       每个partition内部消息强有序，其中的每个消息都有一个序号叫offset 。但是多个partition直接的消息数据是无序的。

3.       因为多个partition之前的消息是无序的，因此kafka不是严格的先进先出，只能说是每个partition内部是严格的先见先出。

4.       一个partition只对应一个broker服务器，一个broker可以管理多个partition。（我们见到的所有集群都这样，跟废话一样）。

5.       消息直接写入文件中，并不存储在内存中

6.       Kafka中的数据会根据时间策略删除（默认是一周），而不是消费完就删除。

7.       kafka集群中各个节点没有主从关系，因此用zookeeper来管理协调kafka集群。因此zookeeper中存着kafka集群的元数据包括所有的Broker节点信息、topic信息。

8.       消费者：消费者自己去zookeeper中维护消费者偏移量。费者可以通过修改偏移量读取任何位置的数据

9.       Topic：一类消息，消息队列。不同的数据写入不同的topic中。例如日志消息写在topic1中，金融类消息写在topic2中。这样的结果是在消费的时候也是有针对的消费某一个topic。

10.   kafka里面的消息是由topic来组成的，简单的我们可以想象为一个队列，一个队列就是一个topic，然后它把每个topic又分为很多个partition，这个是为了做并行的，在每个partition里面是有序的，相当于有序的队列，其中每个消息都有个序号，比如0到12，从前面读往后面写。

11.   这个partition可以很简单想象为一个文件，当数据发过来的时候它就往这个partition上面append，追加就行，kafka和很多消息系统不一样，很多消息系统是消费完了我就把它删掉，而kafka是根据时间策略删除，而不是消费完就删除，在kafka里面没有一个消费完这么个概念，只有过期这样一个概念

12.   Kafka输出的数据是一个kv格式的，因为生产者生产数据的时候有一种方式是hash取模的方式，因此kafka中的数据必定有key值。

13.   重复生产：生产者往kakfa中生产数据的时候，他们之间有一个ack应答机制，回应写入成功，此时会走网络通信，如果此时网络通信了，那么就会造成kafka重复生产问题。这需要修改kakfa源码，在外部无法做到。解决办法是升级kafka版本到0.11版本。

 

第三：kafka的存储、生产消息模型

1.       每个消费者consumer自己去维护消费者偏移量。每个消费者都有对应的group消费者组。

2.       同一个消费者组内，消息只能被消费一次。如果组内的其他人消费过这个消息，我就不能再消费这个消息了。

3.       由2可以得出，在同一个组内，一个partition同一时刻只能被一个消费者来消费。例如kafka集群中有3个partition，group消费者组中有四个消费者。那么就会导致总会有一个消费者处在空闲状态。

4.       不同的组之间的消费者组在同一个partition上消费同一个topic是互相不影响的。

第四：kafka的优点。

1.       Kakfa用到了零拷贝，性能比较好。

2.       Kafka将消息直接写在磁盘中，因此数据不丢失。写（append）的时候是按顺序写的，所以消费的时候也是按照顺序消费的。持久化的同时还保证了顺序读写。

3.       传统拷贝：先将文件拷贝到linux内核的Read buffer，cpu再将Read buffer里面的文件拷贝到Application buffer，cpu再将Application buffer里面的我呢件拷贝到Socket buffer，DMA再将Socket buffer里面的文件拷贝到网络接口的Nic buffer中。

4.       零拷贝：Kafka的零拷贝是指cpu不参加了拷贝，“零拷贝”是指计算机操作的过程中，CPU不需要为数据在内存之间的拷贝消耗资源。而它通常是指计算机在网络上发送文件时，不需要将文件内容拷贝到用户空间（User Space）而直接在内核空间（Kernel Space）中传输到网络的方式。

5.       负载均衡：Kafka是分布式的，因此有数据副本，也就是同一份数据可以到不同的broker上面去，也就是当一份数据，磁盘坏掉的时候，数据不会丢失，比如3个副本，就是在3个机器磁盘都坏掉的情况下数据才会丢。

第五：最终画图结果（图中缺少zookeeper的部分）

 

第六：kafka集群搭建

1.       搭建好之后启动kafka，启动kafka之前要先启动zookeeper集群。

2.       我们在一个配置脚本中取写启动kakfa并将消息保存在哪个文件中。使用命令启动kafka的时候是执行此脚本，同时在命令中需要指定kakfa的集群中的节点、需要指定topic的名称、需要指定partition的个数、需要指定每个partition的副本数。如果有其中一项没指定此命令就会保存。

 

3.       现在kafka集群已经运行起来了，我们看一下集群中的分配情况：此集群中有一个topic名称为t0122的名字，共有三个partition分区（0号、1号、2号），我们以0号分区为例，0号分区所在的leader（其实就是broker服务器）名称是2，副本有三个，存在的节点是2、0、1，检查数据完整性顺序是2、0、1。现在假设0号partition分区中的2号节点(leader：2)挂了，此时要给0号partition分区找一个新的leader，寻找办法就是从副本中按顺序找，所以会找到0号节点。

4.       现在如果1号partition中的0号节点挂掉了（第二行leader=0挂了），按照副本的顺序会找到1号节点。并且第一行中会继续往下找leader，此时他也找到1号节点。最终，1号节点管理3个partition。此时1号节点的负载比较高。

5.       由于上面的负载比较高，因此当我们手动将挂掉的节点重新启动的时候，该节点的leader还会变回原始状态的那个leader。这就是leader均衡机制。此机制是在配置文件中配置的。使用true或者false。

6.       进入zookeeper集群中，我们能看到集群中包含broker节点的信息和topic的信息，这些都是kafka的元数据信息。这些信息被保存在zookeeper的znode目录结构中了。

 

7.       我们还能在zookeeper中找到kafka集群的详细信息（元数据的详细信息）。即:zookeeper的目录文件中保存着zxid等信息，跟我们之前讲zookeeper的时候一致。

 

8.       我们先消费一些数据，然后在zookeeper中查看消费者偏移量（说明消费者偏移量是保存在zookeeper中的consumers消费者目录下）。我们查看的结果是此节点消费者偏移量是0，说明消费的时候，没在此节点上的partition上消费。去其他节点上能找到这个消费者偏移量不是0的节点。

 

9.       删除top0122这个topic，执行结果显示这个topic标记被删除。默认都是一周后会自动删除

 

 

10.   如果我们想把这个topic立马删除，就要先执行delete删除操作，再在kafka集群中挨个节点去删除这个topic目录，最后在zookeeper集群中删除元数据。

 

 

 

总结:

1.       消息：kafka作为分布式消息队列，生产者消费者模式，kafka中的一类消息用topic来表示。

2.       下面我们来说说topic，生产者生产数据的时候，自己决定往哪个topic中的哪个partition去写，根据哈希取模的方式或者轮询的方式，当用哈希取模的方式的时候，产生的消息是kv格式的。Kafka中的一个topic根据这两个方式会把数据分散到多个partition中。

3.       下面我们说说partition，partition是组成topic的单元，集群partition是默认有三个副本的，每个partition由一个leader管理，当前leader对应的broker节点挂掉之后，会根据副本的前后顺序(在zookeeper中存着呢)，让第一个副本作为leader，依次类推（原理：leader会先在zookeeper集群中进行注册，然后生产者作为zookeeper集群的客户端使用watch监听器来监听leader的变更事件）。当我们把原来挂掉的leader节点手动启动之后，这个原来的broker节点又恢复成leader，这就是均衡机制。

4.       下面我们说一下broker。Broker是kafka集群中的节点，kafka集群是无主模式，broker负责消息的读写和存储，消息写在broker节点中的磁盘中，会被追加到log文件的尾部。每条文件在log文件中的位置成为偏移量，偏移量是long型的数字，用来唯一确定一条消息。消息默认保存一周。而由于是无主模式，因此元数据由zookeeper管理。

5.       下面说一下zookeeper，zookeeper集群管理kafka集群中的元数据，元数据包括broker节点的信息和topic的信息（当broker启动后，首先会向zookeeper集群中注册自己的节点信息，在zookeeper中产生znode文件目录，当borker与zookeeper断开连接的时候，该znode目录也会被删除。当一个broker启动时,会向zookeeper注册自己持有的topic和partitions信息,仍然是一个临时znode）。（还可以介绍一下zookeeper攘其外安其内的做法）。

6.       现在元数据和数据都有了，可以介绍一下消费者了。每个消费者自己去维护消费者的偏移量，每个消费者都对应一个消费者组。

7.       下面介绍一下消费者组，同一个topic数据只能被同一个消费者组消费一次，不同消费者组在消费同一个topic的时候互不影响。Kafka集群不是严格模式的先进先出模式，因为同一个topic分布在不同partition中，只能说在本个partition中消息是先进先出的。

8.       网上的资源：搜索：kafka中produce如何把数据存到broker中的，http://www.cnblogs.com/likehua/p/3999538.html

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

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

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

9.    Sparkstreaming+kafka中Driect（迪ruai的）模式的特点：

在介绍Driect模式的时候，需要结合着kafka说：

①    Driect是直接的意思，不像receiver模式一样需要有一个task一直被占用。Receiver模式是kakfa往sparkstreaming中去推数据。driect模式是从sparkstreaming去kafka中拉数据，他把kafka当成hdfs一样。

②   Driect模式的并行度与读取kafka消息队列中topic中的partition个数一致（看kafka）

③   高可用，因为kafka中有3个副本。

④   sparkStreaming自己管理偏移量，它使用checkpoint来更新偏移量。这个过程是异步处理的，它启动两个线程分别执行数据计算与更新偏移量的操作。这样就可能出现这种情况：偏移量更新完了，但是任务提交失败了。这种情况会造成事物不是一致性了（不能保证每条数据恰好被消费一次了）

⑤   如何保证一条数据在sparkStreaming中敲好被消费一次？做法如下：                                当最后一个RDD计算完成之后，在更新偏移量，偏移量一般会放到hbse（把偏移量当做rowkey）或redis或zookeeper中，这就是程序员自己维护偏移量。程序员自己维护偏移量的好处是能保证sparkstreaming从kafka中恰好拉取到一次数据（不会出现④中的情况）。但是sparkstreaming拿到数据之后，有可能在sparkstreaming中处理了多次（因为sparkstreaming有重试机制和推测执行机制），被处理多次就会导致我们的处理结果有重复数据（如保存在了mysql中）。因此我们要保证sparkstreaming中的处理逻辑是幂等操作（例如被打了十下，我只感觉到一次疼。其实就是重复了多次，但我只保存一次）。做幂等操作的时候我们需要引入事物表，拿事物id去检查现在mysql中是否有此事物了。事物id怎么弄？生产者在往kafka中生产数据的时候，给每条数据都带一个id号。因为我kafka中的数据是kv格式的。