初识Kafka

╰+攻爆jí腚メ 2022-08-19 08:57 128阅读 0赞

**（一）Kafka设计背景及原因**

Kafka最初被LinkedIn设计来处理活动流数据(activity stream data)和系统处理数据(operaitonal data)。活动流数据是指像page view、用户搜索关键词等等通过用户操作产生的数据，它的常见场景有时间线(time line)即新鲜事提醒、用户浏览量搜索量排名等等。系统处理数据是服务器性能相关的数据，如CPU、负载、用户请求数等，它的应用场景多数是为后台服务，如在安全方面，可以监控到恶意攻击服务器的用户，从而做出相应措施，还有监控服务器性能，在其出现问题时即时报警等。

这两种数据都属于日志数据的范畴。常见的日志系统，如scribe等都是将这些数据收集起来，然后再通过线下批处理，如hadoop集群等，获取所需的结果。线下处理的频率一般不会太高，比如一个小时甚至一天一次，这是不适合做实时应用的，如timeline这种应用。传统的消息队列系统非常适合这种实时性要求高的场景，但是由于它们都是在内存中维护消息队列，所以处理数据的大小就受到了限制。看到这里，相信聪明的读者已经知晓了Kafka出现的原因，就是要将上面讲的这两种场景整合到一个系统中，即offline的大数据分析和online的实时数据分析都可以通过该系统实现。

Kafka在设计上保留了消息队列的常用操作，而这也使得在其诞生后被越来越广泛的作为一个消息队列使用，而不仅仅局限于处理上面提到的两种数据。

kafka和JMS实现(activeMQ)不同的是:即使消息被消费，消息仍然不会被立即删除.日志文件将会根据broker中的配置要求,保留一定的时间之后删除;比如log文件保留2天,那么两天后,文件会被清除,无论其中的消息是否被消费。kafka通过这种简单的手段,来释放磁盘空间,以及减少消息消费之后对文件内容改动的磁盘IO开支。

**（二）Kafka的设计特色 **

Kafka在设计上有以下几个特色：

1）**消息数据通过磁盘线性存取**

该特性应该是Kafka最让惊叹的特性，使得即使数TB的消息存储也能够保持长时间的稳定性能。在我们的认识中，硬盘较内存的数据处理速度(读写)是慢很多的，所以基本所有设计数据处理的程序都是尽量使用内存。在对消息进行存储和缓存时，Kafka严重地依赖于文件系统。大家普遍认为“磁盘很慢”，因而人们都对持久化结（persistent structure）构能够提供说得过去的性能抱有怀疑态度。实际上，同人们的期望值相比，磁盘可以说是既很慢又很快，这取决于磁盘的使用方式。设计的很好的磁盘结构往往可以和网络一样快。Kafka的设计者在经过一番调研测试后，大胆地采用了全硬盘存取消息数据的方案，他们的主要依据是：

*  硬盘在线性读写的情况下性能优异

在一个由6个7200rpm的SATA硬盘组成的RAID-5磁盘阵列上，线性写入（linear write）的速度大约是300MB/秒，但随即写入却只有50k/秒，其中的差别接近10000倍。线性读取和写入是所有使用模式中最具可预计性的一种方式，因而操作系统采用预读（read-ahead）和后写（write-behind）技术对磁盘读写进行探测并优化后效果也不错。预读就是提前将一个比较大的磁盘块中内容读入内存，后写是将一些较小的逻辑写入操作合并起来组成比较大的物理写入操作。关于这个问题更深入的讨论请参考这篇文章[ACM Queue article][]；实际上他们发现，在某些情况下，顺序磁盘访问能够比随即内存访问还要快！

*  减少JVM的GC触发

JVM中的对象会占用除实际数据外的较多空间（如类的信息等等），Java对象的内存开销（overhead）非常大，往往是对象中存储的数据所占内存的两倍（或更糟） ，浪费空间。而当内存中维护的消息数据逐渐增多后，GC便会被频繁触发，随着堆内数据不断增长而变得越来越不明确，回收所花费的代价也会越来越大，这会极大影响应用的响应速度。因此，舍弃内存作为缓存，使用磁盘可以减少GC触发带来的影响。Kafka论文中与ActiveMQ等消息队列的性能比较也进一步肯定了Kafka的这一设计理念。 而且， 文件系统缓存即使在服务重启之后会仍然保持有效，而不象进程内缓存，进程重启后还需要在内存中进行缓存重建（10G的缓存重建时间可能需要10分钟），否则就需要以一个全空的缓存开始运行（这么做它的初始性能会非常糟糕）。这还大大简化了代码，因为对缓存和文件系统之间的一致性进行维护的所有逻辑现在都是在OS中实现的，这事OS做起来要比我们在进程中做那种一次性的缓存更加高效，准确性也更高。如果你使用磁盘的方式更倾向于线性读取操作，那么随着每次磁盘读取操作，预读就能非常高效使用随后准能用得着的数据填充缓存。

2）**高效吞吐率**

Kafka的设计初衷便是要能处理TB级的数据，即使是在非常普通的硬件上，kafka也可以支持每秒数十万的消息。其更强调的是吞吐率。吞吐率表现在读和写两个方面，Kafka在这两个方面都做了很多优化。

*  写

Kafka在读方面使用了sendfile这个高级系统函数，也即zero-copy技术，感兴趣的同学可以去阅读IBM的文章。这项技术通过减少系统拷贝次数，极大地提高了数据传输的效率。

为了理解sendfile所带来的效果，重要的是要理解将数据从文件传输到socket的数据路径：

1.  操作系统将数据从磁盘中读取到内核空间里的页面缓存
2.  应用程序将数据从内核空间读入到用户空间的缓冲区
3.  应用程序将读到的数据写回内核空间并放入socke的缓冲区
4.  操作系统将数据从socket的缓冲区拷贝到NIC（网络借口卡，即网卡）的缓冲区，自此数据才能通过网络发送出去

这样效率显然很低，因为里面涉及4次拷贝，2次系统调用。使用sendfile就可以避免这些重复的拷贝操作，让OS直接将数据从页面缓存发送到网络中，其中只需最后一步中的将数据拷贝到NIC的缓冲区。

我们预期的一种常见的用例是一个话题拥有多个消息使用者。采用前文所述的零拷贝优化方案，数据只需拷贝到页面缓存中一次，然后每次发送给使用者时都对它进行重复使用即可，而无须先保存到内存中，然后在阅读该消息时每次都需要将其拷贝到内核空间中。如此一来，消息使用的速度就能接近网络连接的极限。

要得到Java中对send'file和零拷贝的支持方面的更多背景知识，请参考IBM developerworks上的这篇[文章][Link 1]。

*  读

多数情况下系统的瓶颈是网络而不是CPU。这一点对于需要将消息在个数据中心间进行传输的数据管道来说，尤其如此。当然，无需来自Kafka的支持，用户总是可以自行将消息压缩后进行传输，但这么做的压缩率会非常低，因为不同的消息里都有很多重复性的内容（比如JSON里的字段名、web日志中的用户代理或者常用的字符串）。高效压缩需要将多条消息一起进行压缩而不是分别压缩每条消息。理想情况下，以端到端的方式这么做是行得通的 —— 也即，数据在消息生产者发送之前先压缩一下，然后在服务器上一直保存压缩状态，只有到最终的消息使用者那里才需要将其解压缩。

通过运行递归消息集，Kafka对这种压缩方式提供了支持。一批消息可以打包到一起进行压缩，然后以这种形式发送给服务器。这批消息都会被发送给同一个消息使用者，并会在到达使用者那里之前一直保持为被压缩的形式。

Kafka支持GZIP和Snappy压缩协议。关于压缩的更多更详细的信息，请参见[这里][Link 2]。

**3）消费状态由消费者自己维护**

许多MQ都具备一个确认功能，当消息发出后仅把它标示为**已发送**而不是**已使用**，然后代理需要等到来自使用者的特定的确认信息后才将消息记录为**已使用**。这种策略的确解决了丢失消息的问题，但由此产生了新问题。首先，如果使用者已经处理了该消息但却未能发送出确认信息，那么就会让这一条消息被处理两次。第二个问题是关于性能的，这种策略中的代理必须为每条单个的消息维护多个状态（首先为了防止重复发送就要将消息锁定，然后，然后还要将消息标示为已使用后才能删除该消息）。另外还有一些棘手的问题需要处理，比如，对于那些以发出却未得到确认的消息该如何处理？

对于消息传递的语义，系统可以提供的几种可能的消息传递保障如下所示:

*  最多一次—这种用于处理前段文字所述的第一种情况。消息在发出后立即标示为已使用，因此消息不会被发出去两次，但这在许多故障中都会导致消息丢失。 
 *  至少一次—这种用于处理前文所述的第二种情况，系统保证每条消息至少会发送一次，但在有故障的情况下可能会导致重复发送。 
 *  仅仅一次—这种是人们实际想要的，每条消息只会而且仅会发送一次。

这个问题已得到广泛的研究，属于“事务提交”问题的一个变种。提供仅仅一次语义的算法已经有了，两阶段或者三阶段提交法以及Paxos算法的一些变种就是其中的一些例子，但它们都有与生俱来的的缺陷。这些算法往往需要多个网络往返（round trip），可能也无法很好的保证其活性（liveness）（它们可能会导致无限期停机）。FLP结果给出了这些算法的一些基本的局限。

Kafka对元数据做了两件很不寻常的事情：

1.  代理（broker）将数据流划分为一组互相独立的分区。这些分区的语义由生产者（producer）定义，由生产者来指定每条消息属于哪个分区。一个分区内的消息以到达代理的时间为准进行排序，将来按此顺序将消息发送给使用者。这么一来，就用不着为每一条消息保存一条元数据（比如说，将消息标示为已使用）了，我们只需为使用者、话题和分区的每种组合记录一个“最高水位标记”（high water mark）即可。因此，标示使用者状态所需的元数据总量实际上特别小。在Kafka中，我们将该最高水位标记称为“偏移量”（offset）。
2.  在Kafka中，使用者（consumer）将消息状态信息保存到它们的消息处理节点的那个数据存储（datastore）中。这样就消除了分布式的部分，从而解决了分布式中的一致性问题！这在非事务性系统中也有类似的技巧可用。搜索系统可用将使用者状态信息同它的索引段（index segment）存储到一起。尽管这么做可能无法保证数据的持久性（durability），但却可用让索引同使用者状态信息保存同步：如果由于宕机造成有一些没有刷新到磁盘的索引段信息丢了，我们总是可用从上次建立检查点（checkpoint）的偏移量处继续对索引进行处理。与此类似，Hadoop的加载作业（load job）从Kafka中并行加载，也有相同的技巧可用。每个Mapper在map任务结束前，将它使用的最后一个消息的偏移量存入HDFS。这个决策还带来一个额外的好处。使用者可用故意*回退（rewind）*到以前的偏移量处，再次使用一遍以前使用过的数据。虽然这么做违背了队列的一般协约（contract），但对很多使用者来讲却是个很基本的功能。举个例子，如果使用者的代码里有个Bug，而且是在它处理完一些消息之后才被发现的，那么当把Bug改正后，使用者还有机会重新处理一遍那些消息。

**4）分布式**

Kafka通常情况下是运行在集群中的服务器上。没有中央的“主”节点。代理彼此之间是对等的，不需要任何手动配置即可可随时添加和删除。同样，生产者和消费者可以在任何时候开启。每个代理都可以在Zookeeper(分布式协调系统)中注册的一些元数据（例如，可用的主题）。生产者和消费者可以使用Zookeeper发现主题和相互协调。

*  broker的部署是一种no central master的概念，并且每个节点都是同等的，节点的增加和减少都不需要改变任何配置。 
 *  producer和consumer通过zookeeper去发现topic，并且通过zookeeper来协调生产和消费的过程。 
 *  producer、consumer和broker均采用TCP连接，通信基于NIO实现。Producer和consumer能自动检测broker的增加和减少。

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

**5）异步发送**

对于可伸缩的消息系统而言，异步非阻塞式操作是不可或缺的。在Kafka中，生产者有个选项（producer.type=async）可用指定使用异步分发出产请求（produce request）。这样就允许用一个内存队列（in-memory queue）把生产请求放入缓冲区，然后再以某个时间间隔或者事先配置好的批量大小将数据批量发送出去。因为一般来说数据会从一组以不同的数据速度生产数据的异构的机器中发布出，所以对于代理而言，这种异步缓冲的方式有助于产生均匀一致的流量，因而会有更佳的网络利用率和更高的吞吐量。

**6）支持Hadoop并行数据加载**

具有伸缩性的持久化方案使得Kafka可支持批量数据装载，能够周期性将快照数据载入进行批量处理的离线系统。我们利用这个功能将数据载入我们的数据仓库（data warehouse）和Hadoop集群。

批量处理始于数据载入阶段，然后进入非循环图（acyclic graph）处理过程以及输出阶段（支持情况在[这里][Link 3]）。支持这种处理模型的一个重要特性是，要有重新装载从某个时间点开始的数据的能力（以防处理中有任何错误发生）。对于Hadoop，我们通过在单个的map任务之上分割装载任务对数据的装载进行了并行化处理，分割时，所有节点/话题/分区的每种组合都要分出一个来。Hadoop提供了任务管理，失败的任务可以重头再来，不存在数据被重复的危险。

**（三）实现原理**

1）术语

*  Topic：一个Topic可以认为是一类消息,每个topic将被分成多个partition(区),每个partition在存储层面是append log文件.任何发布到此partition的消息都会直接追加到log文件的尾部,每条消息在文件中的位置称为offset(偏移量),offset为一个long型数字,它唯一的标记一条消息.kafka并没有提供其他额外的索引机制来存储offset,因为在kafka中几乎不允许对消息进行"随机读-写"。不同的数据可以按照不同的topic存储。
 *  Message：由消息生产者（producer）发布关于某话题（topic）的消息。在kafka中，消息是被发布到broker的topic中。而consumer也是从相应的topic中拿数据。也就是说，message是按topic存储的。
 *  Partition：同一个topic下可以设置多个partition，目的是为了提高并行处理的能力。可以将同一个topic下的message存储到不同的partition下。
 *  Offset：kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka。
 *  Producers：将消息发布到指定的Topic中,同时Producer也能决定将此消息归属于哪个partition;比如基于"round-robin"方式或者通过其他的一些算法等。
 *  Consumer Group：生产者（producer）、消费者（consumer）和代理（broker）都可以运行在作为一个逻辑单位的、进行相互协作的集群中不同的机器上。对于代理和生产者，这么做非常自然，但使用者却需要一些特殊的支持。每个消费者进程都属于一个*使用者小组（consumer group）* 。准确地讲，每条消息都只会发送给每个使用者小组中的一个进程。因此，使用者小组使得许多进程或多台机器在逻辑上作为一个单个的使用者出现。使用者小组这个概念非常强大，可以用来支持JMS中*队列（queue）*或者*话题（topic）*这两种语义。
 *   *  如果所有的consumer都具有相同的group,这种情况和P2P模式的Queue模式很像;消息将会在consumers之间负载均衡。
     *  如果所有的consumer都具有不同的group, 这种情况和Pub/Sub模式的Topic模式很像 ;消息将会广播给所有的消费者。
 *  Distribution：  一个Topic的多个partitions,被分布在kafka集群中的多个server上;每个server(kafka实例)负责partitions中消息的读写操作;此外kafka还可以配置partitions需要备份的个数(replicas),每个partition将会被备份到多台机器上,以提高可用性。  基于replicated方案,那么就意味着需要对多个备份进行调度;每个partition都有一个server为"leader";leader负责所有的读写操作,如果leader失效,那么将会有其他follower来接管(成为新的leader);follower只是单调的和leader跟进,同步消息即可..由此可见作为leader的server承载了全部的请求压力,因此从集群的整体考虑,有多少个partitions就意味着有多少个"leader",kafka会将"leader"均衡的分散在每个实例上,来确保整体的性能稳定。

2）部署架构

多个broker协同合作，producer和consumer部署在各个业务逻辑中被频繁的调用，三者通过zookeeper管理协调请求和转发。producer到broker的过程是push，也就是有数据就推送到broker，而consumer到broker的过程是pull，是通过consumer主动去拉数据的，而不是broker把数据主动发送到consumer端的。

![初识Kafka][Kafka]

3）运行架构

![初识Kafka][Kafka]

假设broker的数量只有一台，并按照上图进行部署：

*  Server-1 broker其实就是kafka的server，因为producer和consumer都要去连它。Broker主要还是做存储用。 
 *  Server-2是zookeeper的server端，zookeeper负责集群调度，它维持了一张表，记录了各个节点的IP、端口等信息（以后还会讲到，它里面还存了kafka的相关信息）。 
 *  Server-3、4、5他们的共同之处就是都配置了zkClient，更明确的说，就是运行前必须配置zookeeper的地址，道理也很简单，这之间的连接都是需要zookeeper来进行分发的。 
 *  Server-1和Server-2的关系，他们可以放在一台机器上，也可以分开放，zookeeper也可以配集群。目的是防止某一台挂了。

简单说下整个系统运行的顺序：

1.  启动zookeeper的server 
2.  启动kafka的server 
3.  Producer如果生产了数据，会先通过zookeeper找到broker，然后将数据存放进broker 
4.  Consumer如果要消费数据，会先通过zookeeper找对应的broker，然后消费。

**4）逻辑架构**

Topic、Partition、Replication关系

*  在Kafka中，一个Topci可以分成若干个Partition，并尝试均匀的存储在可用的Broker集群中。可以对每个Broker和每个Topic设置 Partition数量 ，每个Partition都会分配一个ID。每个Partition又可以分成若干个Replication备份在其他可用的Broker集群中，但每个Broker只能为每个Partition存放一个Replication，Replication的ID就是该宿主Broker的ID。例如一个Topic的Partition=4，Replica=3，那么Kafka将会为每个Partition创建3个Replica，一共12个，每个Partition的Replica都会选择一个Broker作为Leader。
 *  默认情况下，生产消息和消费消息的模式为Sync同步模式，这样生产者使用send()方法提交写入blocks将会等到在所有可用的Broker中完成Replication，这样可以保证消息不会丢失，生产架构推荐 Replica=3。如果对延迟敏感的生产者，那么可以调整写入blocks会Async异步模式，或者只写入Broker Leader，那么这样做的后果就是放弃消息可靠性，因为消费者只能消费他们提交的消息，而看不见备份的消息。
 *  对于高容量的Topic，每个Broker都会配置超过1个Partition。 Partition的增加将会增加Write写操作的I/O并非，同时也会增加消费者Consumer的并发度，因为Partition是分布式消息的处理单元。也就是说越多的Partition将会增加：a）产生越多的文件和打开越多的文件句柄；b）更多的偏移量offset需要被Consumer确认，从而增加Zookeeper的负载；

创建Topic的方式

*  为Broker配置auto.create.topics.enable，当Broker接收到该Topic的第一个消息时，将通过num.partitions和default.replication.factor创建Topic
 *  使用bin/kafka-create-topic.sh和bin/kafka-list-topic.sh命令创建和查看Topic

**（四）Client API**

1）kafka.javaapi.producer.Producer

*  public void send(KeyedMessage message);  向单个主题Topic发送消息，通过Key键分区，同时支持同步、异步生产者 
 *  public void send(KeyedMessage message);  向多个主题Topic发送消息 
 *  public void close();  向所有的Kafka Brokers关闭生产者Producer连接池

2）High Level Consumer API

*  kafka.consumer.Consumer 
 *   *  public static kafka.javaapi.consumer.ConsumerConnector createJavaConsumerConnector(config: ConsumerConfig); 创建一个ConsumerConnector 
 *   kafka.javaapi.consumer.ConsumerConnector

*  public Map>> createMessageStreams(Map topicCountMap, Decoder keyDecoder,  Decoder valueDecoder); 根据泛型，为每个主题Topic创建一个消息流集合 
 *  public Map>> createMessageStreams(Map Integer> topicCountMap); 根据默认编码decoder，为每个主题Topic创建一个默认的消息流集合 
 *  public List> createMessageStreamsByFilter(TopicFilter topicFilter, int numStreams, Decoder   
 *  keyDecoder, Decoder valueDecoder); 根据泛型，以及所匹配的通配符和返回的消息流数目， 返回对应的消息流集合 
 *  public List> createMessageStreamsByFilter(TopicFiltertopicFilter, int numStreams); 根据默认编码decoder，以及所匹配的通配符和返回的消息流数目，返回对应的消息流集合 
 *  public List> createMessageStreamsByFilter(TopicFiltertopicFilter); 根据默认编码decoder，以及所匹配的通配符，返回默认的消息流集合 
 *  public void commitOffsets(); 向该connector所连接的所有主题Topic或者分区Partition提交偏移量 Offset 
 *  public void shutdown(); 关闭该connector

3）Low Level Simple Consumer API

*  kafka.javaapi.consumer.SimpleConsumer 
 *   *  public FetchResponse fetch(request: kafka.javaapi.FetchRequest); 从一个主题Topic获取消息 
     *  public kafka.javaapi.TopicMetadataResponse send(request: kafka.javaapi.TopicMetadataRequest); 获取一组主题Topic的元数据 
     *  public kafka.javaapi.OffsetResponse getOffsetsBefore(request: OffsetRequest); 根据给定时间，获取有效的偏移量 
     *  public void close(); 关闭该SimpleConsumer

通常情况， High Level Consumer API 已经足够使用，特殊场景（如：Consumer重启，重新初始化偏移量可以使用 Low Level Simple Consumer API ）

4）Kafka Hadoop Consumer API

[ https://github.com/apache/kafka/tree/trunk/contrib/hadoop-consumer][https_github.com_apache_kafka_tree_trunk_contrib_hadoop-consumer]

**（五）安装部署运行**

*  安装

$  tar xzf kafka-.tgz

$  cd kafka-

$  ./sbt update

$  ./sbt package

$  ./sbt assembly-package-dependency

默认情况下，kafka-0.8.0使用scala 2.8.0编译，若要使用scala其他版本，如2.9.2可使用如下命令：

$  ./sbt "++2.9.2 update"

$  ./sbt "++2.9.2 package"

$  ./sbt "++2.9.2 assembly-package-dependency"

*  启动broker，前提zookeeper已启动

**1）单节点单broker**

后台启动：>nohup  ./ kafka-server-start.sh ../config/server.properties >/dev/null 2>&1 &

**2）单节点多broker**

修改%KAFKA\_HOME%/config的server.properties

$  cp config/server.properties config/server-1.properties

$  cp config/server.properties config/server-2.properties

$  mv config/server.properties config/server-3.properties

内容修改如下：

vim config/server-1.properties

broker.id=1

port=9092

host.name=172.16.38.220

log.dir=/tmp/kafka-logs-1

zookeeper.connect=172.16.38.220:2181,172.16.38.220:2182,172.16.38.220:2183/kafka

vim  config/server-2.properties

broker.id=2

port=9093

host.name=172.16.38.220

log.dir=/tmp/kafka-logs-2

zookeeper.connect=172.16.38.220:2181,172.16.38.220:2182,172.16.38.220:2183/kafka

vim  config/server-3.properties

broker.id=3

port=9094

host.name=172.16.38.220

log.dir=/tmp/kafka-logs-3

zookeeper.connect=172.16.38.220:2181,172.16.38.220:2182,172.16.38.220:2183/kafka

**先在Zookeeper建立Kafka根目录/kafka，再启动后台实例：**

$export JMX\_PORT=9997

$nohup  bin/ kafka-server-start.sh config/server-1.properties >/dev/null 2>&1 &

$nohup  bin/ kafka-server-start.sh config/server-1.properties >logs/nohup.server1.out 2>&1 &

$ export JMX\_PORT=9998

$nohup  bin / kafka-server-start.sh config/server-2.properties >/dev/null 2>&1 &

$nohup  bin / kafka-server-start.sh config/server-2.properties >logs/nohup.server2.out 2>&1 &

$ export JMX\_PORT=9999

$nohup  bin / kafka-server-start.sh config/server-3.properties >/dev/null 2>&1 &

$nohup  bin / kafka-server-start.sh config/server-3.properties >logs/nohup.server3.out 2>&1 &

*  创建主题

bin/kafka-create-topic.sh --zookeeper      172.16.38.220: 2181/kafka  \--partition 3 --replica 3 --topic zerg. hydra

*  查看主题：

bin/kafka-list-topic.sh --zookeeper 172.16.38.220:2181/kafka

1.  partition：该topic的分区
2.  leader：该partition的leader，负责处理所有的读、写操作，每个broker节点都会成为随机分配partition的leader
3.  replicas：该partition的副本所在节点列表，无论该节点是否可用
4.  isr：该partition的副本同步的节点列表，这是副本节点列表的子集，显示了目前跟随leader还可用的节点信息

*  创建消息

先关联主题： bin/kafka-console-producer.sh --broker-list    172.16.38.220 :909 3,   172.16.38.220 :9094 --sync  \--topic zerg.hydra

再输入测试消息：Hello world

*  接收消息

从开始接收消息：bin/kafka-console-consumer.sh --zookeeper  172.16.38.220:2181/kafka  \--topic  zerg. hydra  \--from-beginning

观察收到消息：Hello world

*  测试容错

先关联主题： bin/kafka-console-producer.sh --broker-list   172.16.38.220 :909 3,  172.16.38.220 :9094 --sync  \--topic zerg.hydra

输入测试消息：Hi! R eplicated message!

杀死一个进程： pkill -9 -f server-1.properties

再次查看主题：bin/kafka-list-topic.sh --zookeeper 172.16.38.220:2181/kafka

再次接收消息：bin/kafka-console-consumer.sh --zookeeper  172.16.38.220:2181/kafka  \--topic  zerg. hydra  \--from-beginning

*  删除主题

a)关闭Kafka集群

pkill -9 -f server-1.properties

pkill -9 -f server-2.properties

pkill -9 -f server-3.properties

b)删除kafka日志文件

rm -rf /tmp/kafka-logs-\*  
 c)删除zookeeper数据   
d)重启Kafka集群

**（六）开发测试**

1）搭建Eclipse本地开发环境

[ https://cwiki.apache.org/confluence/display/KAFKA/Developer+Setup][https_cwiki.apache.org_confluence_display_KAFKA_Developer_Setup]

按照文档，根据自身Eclipse版本选择下载Scala插件，这里选择为：

Scala IDE 3.0.0 Release Eclipse 3.7 (Indigo) For Scala 2.9.x   [http://download.scala-ide.org/sdk/e37/scala29/stable/site][http_download.scala-ide.org_sdk_e37_scala29_stable_site]

这里设置方式以Eclipse setup (with SBT eclipse plugin)配置为主，但需要参考Eclipse setup (without SBT eclipse plugin)的方式添加 Scala Library，并且对于contrib（hadoop-consumer和hadoop-producer）、examples（kafka-java-examples）、perf（kafka-perf）都需要对它们的target目录下的ivy-0.8.0.xml删除kafka\_2.8.0的依赖，导入完成后进行Scala Unit Test测试。

设置Debug：

[ http://scala-ide.org/docs/current-user-doc/features/scaladebugger/index.html][http_scala-ide.org_docs_current-user-doc_features_scaladebugger_index.html]

**（七）管理工具**

1）控制Shutdown

当一个Broker需要停机时，需要有效的将其所管理的Partition交由副本的Leader进行接管。否则在此期间，读写客户端都不能操作该Partition，直到选举完成，而选举是需要对Zookeeper进行读写，这样就存在一些延时。因此，高效的Shutdown，应该是主动的移交Partition管理权，减少不可用时间。两种方法可以办到：

*  设置controlled.shutdown.enable=true，那么Broker在shundown之前，会根据controlled.shutdown.max.retries 和controlled.shutdown.retry.backoff.ms的配置情况，尝试至多交权次数和尝试交权时间间隔
 *  使用命令行工具./kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper 172.16.38.220:2181/ --broker 2 --num.retries 3 --retry.interval.ms 60。注意：命令行工具并不会终结Broker进程，而只是将其所负责的Partition交由其他Broker管理。但由于 Controller 已经Shutdown，因此虽然Kafka的Broker进程还在，依旧无法正常使用，必须收到

当ack=0时，客户端Client（Producer/Consumer）将会从old broker leader返回一个错误，这时它们会从Zookeeper刷新Partition的Metedata获取一个新的leader，然后再次尝试生产/消费消息。由于ack=0，Producer并不会等待broker的确认，那么以外着这将丢失数据，直到这个shutdown的leader重新成为该Partition的Leader。

如果该Partition没有一个可用的副本集合，那么该工具并不会shutdown leader，以免数据丢失。

2）控制Preferred Replica Leader Tool选举

一般情况下，每个Partition都会有多个副本Replica，默认情况该Partition的Broker就是其副本的Preferred Replica。这样可以保证在整个集群中，Partition和Broker相对平衡，但由于一些故障、Controller Shutdown等原因，将打破平衡。而Preferred Replica Leader Election Tool工具正是用于对Replica Leader的均衡选举工作。

使用命令行工具. /kafka-preferred-replica-election.sh --zookeeper 172.16.38.220:2181/ --path-to-json-file topicPartitionList.json

这里zookeeper是必须的，而topic的partition选举json数据是可选的，如果没有，那么该工具会从zookeeper查询所有topic的所有partition进行选举

3)显示主题

./kafka-list-topic.sh --zookeeper 172.16.38.220:2181

可选参数：--under-replicated-partitions 只显示拥有副本的Partition   \--unavailable-partitions只显示不可用的Partitio

4）创建主题

./kafka-create-topic.sh --zookeeper  172.16.38.220  :2181  \--partition 3 --replica 3 --topic zerg. hydra

如果Broker的auto.create.topics.enable=true，Kafka将会在默认的Partition和Replica上创建主题，当然也可以通过可选参数指定配置

5）增加分区

为了提高并发，Kafka会将Topic的消息通过Partition分布到多台服务器进行读写，以提高性能。有时候分区需要进行扩展，那么就可以使用该工具

./kafka-add-partitions.sh   \--zookeeper  172.16.38.220  :2181  \--topic zerg. hydra  \--partition 3

6）重新分配分区Reassign Partitions Tool

跟Preferred Replica Leader Tool类似， Reassign Partitions Tool也用于是集群更加平衡，均衡的处理分区的读写。

./kafka-reassign-partitions.sh --zookeeper 172.16.38.220:2181 --broker-list '1,2,3' --status-check-json-file topics-to-move.json --topics-to-move-json-file topics-to-move.json --execute

7）状态日志收集整理 StateChangeLogMerger Tool

每个在Kafka集群中的Broker都会发出状态变更的日志， StateChangeLogMerger  可以收集、过滤、合并一个集群全局的状态变更日志，以便分析查找问题。

./kafka-run-class.sh kafka.tools.StateChangeLogMerger --logs-regex /apps/programs/kafka-0.8.0-src/bin/logs/state-change.log\* --start-time '2013-12-20 14:00:00,000'

**（八）性能参考**

[ https://cwiki.apache.org/confluence/display/KAFKA/Performance+testing][https_cwiki.apache.org_confluence_display_KAFKA_Performance_testing]

[ http://blog.liveramp.com/2013/04/08/kafka-0-8-producer-performance-2/][http_blog.liveramp.com_2013_04_08_kafka-0-8-producer-performance-2]

./kafka-producer-perf-test.sh --broker-list=172.16.38.219:9092,172.16.38.219:9093,172.16.38.219:9094 --csv-reporter-enabled=true --metrics-dir=/apps/programs/kafka-0.8.0-src/bin/reports --compression-codec=2 --message-size=200 --messages 10000000 --topic perf-test1 --request-num-acks=-1 --batch-size 200 --threads 1

结论：

Producer的增加，并不会显著的增加吞吐量；

Topic的增加，并不会显著的增加吞吐量；

Broker或者Disk的增加，以致Partition的增加，起初会显著提高吞吐量；但后期增加提升效果不再线性增长；

Replica的增加，对吞吐量的降低幅度较小；

当Broker集群的吞吐量未达到上限时，延迟时间相对均衡；

当Broker集群的吞吐量达到上限时，如果Producer继续生产消息，即时消息量很少，也会出现很大延迟；

整个架构设计，增加Broker集群数量，比多线程执行Producer更加有效；

**（九）重要配置**

*  Borker：

broker.id broker唯一标识，数字即可

log.dirs 数据文件保存路径，默认值/tmp/kafka-logs

zookeeper.connect zk集群，如172.16.38.220:2181,172.16.38.220:2182,172.16.38.220:2183/kafka

message.max.bytes 接收消息最大限制，默认值1000000

num.io.threads 用于处理磁盘I/O线程数，至少要多于磁盘个数，默认值8

num.partitions 每个topic的分区个数，默认值为1

default.replication.factor 每个topic分区partition的副本个数，默认值为1

zookeeper.session.timeout.ms zk心跳检测时间，默认值6000

zookeeper.connection.timeout.ms zk连接超时时间，默认值6000

*  Consumer：

group.id 标识consumer所属的group，将多个consumer标识相同的groupId，可以获得P2P的语义，多个consumer标识不同的groupId可以获得Pub/Sub的语义

zookeeper.connect zk集群，如172.16.38.220:2181,172.16.38.220:2182,172.16.38.220:2183/kafka

consumer.id consumer本身标识，如无设置，自动生成，默认值为null

fetch.message.max.bytes 抓取消息的最大限制，建议至少要大于Broker可允许的消息大小

auto.commit.enable 向zk自动提交消费消息偏移量，默认值为true

auto.commit.interval.ms 向zk自动提交消费消息偏移量间隔时间，默认时间60\*1000

queued.max.message.chunks 最大消息缓冲区大小，每个chunk都可以达到fetch.message.max.bytes的上限，默认值为10

client.id 用于跟踪每次请求调用情况，默认值为group.id

zookeeper.session.timeout.ms zk心跳检测时间，默认值6000

zookeeper.connection.timeout.ms zk连接超时时间，默认值6000

*  Producer：

metadata.broker.list broker列表，如172.16.38.219:9093,172.16.38.219:9094

request.required.acks 应答机制，0：producer从不等待broker的应答；1：producer等待broker的应答；-1：producer 等待所有broker replica同步完成消息处理后的应答

producer.type 消息发送机制，sync：同步模式；async：后台进程异步模式。async可以批量打包处理，提高吞吐量，但同时有可能丢失数据

serializer.class 消息序列化，默认kafka.serializer.DefaultEncoder将以byte\[\]数组进行编码

key.serializer.class 消息键值，用于发送KeyedMessage消息，若无设置将于消息本身一样

partitioner.class 分区配置，默认kafka.producer.DefaultPartitioner，基于key的哈希算法

compression.codec 消息压缩算法，none、gzip、snappy

message.send.max.retries 尝试不断发送消息的最多次数

retry.backoff.ms 每次尝试前刷新metadata寻找broker lead的时间

batch.num.messages 在async模式，批量处理的消息个数

queue.buffering.max.ms 在async模式，最长缓冲发送时间

queue.buffering.max.messages 在async模式，最多在缓冲队列保留的消息个数

**（十）总结**

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

[http://www.oschina.net/translate/kafka-design?lang=chs][http_www.oschina.net_translate_kafka-design_lang_chs]

[ACM Queue article]: http://deliveryimages.acm.org/10.1145/1570000/1563874/jacobs3.jpg
[Link 1]: http://www.ibm.com/developerworks/linux/library/j-zerocopy
[Link 2]: https://cwiki.apache.org/confluence/display/KAFKA/Compression
[Link 3]: http://kafka.apache.org/azkaban
[Kafka]: 
[https_github.com_apache_kafka_tree_trunk_contrib_hadoop-consumer]: https://github.com/apache/kafka/tree/trunk/contrib/hadoop-consumer
[https_cwiki.apache.org_confluence_display_KAFKA_Developer_Setup]: https://cwiki.apache.org/confluence/display/KAFKA/Developer+Setup
[http_download.scala-ide.org_sdk_e37_scala29_stable_site]: http://download.scala-ide.org/sdk/e37/scala29/stable/site
[http_scala-ide.org_docs_current-user-doc_features_scaladebugger_index.html]: http://scala-ide.org/docs/current-user-doc/features/scaladebugger/index.html
[https_cwiki.apache.org_confluence_display_KAFKA_Performance_testing]: https://cwiki.apache.org/confluence/display/KAFKA/Performance+testing
[http_blog.liveramp.com_2013_04_08_kafka-0-8-producer-performance-2]: http://blog.liveramp.com/2013/04/08/kafka-0-8-producer-performance-2/
[http_www.oschina.net_translate_kafka-design_lang_chs]: http://www.oschina.net/translate/kafka-design?lang=chs