kafka查看topic中的数据

如何决定kafka集群中话题的分区的数量

如何决定kafka集群中topic,partition的数量，这是许多kafka用户经常遇到的问题。本文列举阐述几个重要的决定因素，以提供一些参考。
分区多吞吐量更高

一个话题topic的各个分区partiton之间是并行的。在producer和broker方面，写不同的分区是完全并行的。因此一些昂贵的操作比如压缩，可以获得更多的资源，因为有多个进程。在consumer方面，一个分区的数据可以由一个consumer线程在拉去数据。分区多，并行的consumer（同一个消费组）也可以多。因此通常，分区越多吞吐量越高。
基于吞吐量可以获得一个粗略的计算公式。先测量得到在只有一个分区的情况下，Producer的吞吐量(P)和Consumer的吞吐量(C)。那如果总的目标吞吐量是T的话，max(T/P,T/C)就是需要的最小分区数。在单分区的情况下，Producer的吞吐量可以通过一些配置参数，比如bath的大小、副本的数量、压缩格式、ack类型来测得。而Consumer的吞吐量通常取决于应用程序处理每一天消息逻辑。这些都是需要切合实际测量。
随着时间推移数据量的增长可能会需要增加分区。有一点需要注意的是，Producer者发布消息通过key取哈希后映射分发到一个指定的分区，当分区数发生变化后，会带来key和分区映射关系发生变化。可能某些应用程序依赖key和分区映射关系，映射关系变化了，程序就需要做相应的调整。为了避免这种key和分区关系带来的应用程序修改。所以在分区的时候尽量提前考虑，未来一年或两年的对分区数据量的要求。
除了吞吐量，还有一些其他的因素，在定分区的数目时是值得考虑的。在某些情况下，太多的分区也可能会产生负面影响。

分区多需要的打开的文件句柄也多

每个分区都映射到broker上的一个目录，每个log片段都会有两个文件（一个是索引文件，另一个是实际的数据文件）。分区越多所需要的文件句柄也就越多，可以通过配置操作系统的参数增加打开文件句柄数。

分区多增加了不可用风险

kafka支持主备复制，具备更高的可用性和持久性。一个分区（partition）可以有多个副本，这些副本保存在不同的broker上。每个分区的副本中都会有一个作为Leader。当一个broker失败时，Leader在这台broker上的分区都会变得不可用，kafka会自动移除Leader，再其他副本中选一个作为新的Leader。Producer和Consumer都只会与Leader相连。
一般情况下，当一个broker被正常关机时，controller主动地将Leader从正在关机的broker上移除。移动一个Leader只需要几毫秒。然当broker出现异常导致关机时，不可用会与分区数成正比。假设一个boker上有2000个分区，每个分区有2个副本，那这样一个boker大约有1000个Leader，当boker异常宕机，会同时有1000个分区变得不可用。假设恢复一个分区需要5ms，1000个分区就要5s。
分区越多，在broker异常宕机的情况，恢复所需时间会越长，不可用风险会增加。

分区多会增加点到点的延迟

这个延迟需要体现在两个boker间主备数据同步。在默认情况下，两个boker只有一个线程负责数据的复制。

根据经验，每个boker上的分区限制在100*b*r内（b指集群内boker的数量，r指副本数量）。
分区多会增加客户端的内存消耗

kafka0.8.2后有个比较好的特色，新的Producer可以允许用户设置一个缓冲区，缓存一定量的数据。当缓冲区数据到达设定量或者到时间，数据会从缓存区删除发往broker。如果分区很多，每个分区都缓存一定量的数据量在缓冲区，很可能会占用大量的内存，甚至超过系统内存。
Consumer也存在同样的问题，会从每个分区拉一批数据回来，分区越多，所需内存也就越大。
根据经验，应该给每个分区分配至少几十KB的内存。
总结

在通常情况下，增加分区可以提供kafka集群的吞吐量。然而，也应该意识到集群的总分区数或是单台服务器上的分区数过多，会增加不可用及延迟的风险。

如何确定Kafka的分区数，key和consumer线程数，以及不消费问题解决

在Kafak中国社区的qq群中，这个问题被提及的比例是相当高的，这也是Kafka用户最常碰到的问题之一。本文结合Kafka源码试图对该问题相关的因素进行探讨。希望对大家有所帮助。
怎么确定分区数？
“我应该选择几个分区？”——如果你在Kafka中国社区的群里，这样的问题你会经常碰到的。不过有些遗憾的是，我们似乎并没有很权威的答案能够解答这样的问题。其实这也不奇怪，毕竟这样的问题通常都是没有固定答案的。Kafka官网上标榜自己是"high-throughput distributed messaging system"，即一个高吞吐量的分布式消息引擎。那么怎么达到高吞吐量呢？Kafka在底层摒弃了Java堆缓存机制，采用了操作系统级别的页缓存，同时将随机写操作改为顺序写，再结合Zero-Copy的特性极大地改善了IO性能。但是，这只是一个方面，毕竟单机优化的能力是有上限的。如何通过水平扩展甚至是线性扩展来进一步提升吞吐量呢？ Kafka就是使用了分区(partition)，通过将topic的消息打散到多个分区并分布保存在不同的broker上实现了消息处理(不管是producer还是consumer)的高吞吐量。
Kafka的生产者和消费者都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此分区实际上是调优Kafka并行度的最小单元。对于producer而言，它实际上是用多个线程并发地向不同分区所在的broker发起Socket连接同时给这些分区发送消息；而consumer呢，同一个消费组内的所有consumer线程都被指定topic的某一个分区进行消费(具体如何确定consumer线程数目我们后面会详细说明)。所以说，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。
但分区是否越多越好呢？显然也不是，因为每个分区都有自己的开销：
一、客户端/服务器端需要使用的内存就越多
先说说客户端的情况。Kafka 0.8.2之后推出了Java版的全新的producer，这个producer有个参数batch.size，默认是16KB。它会为每个分区缓存消息，一旦满了就打包将消息批量发出。看上去这是个能够提升性能的设计。不过很显然，因为这个参数是分区级别的，如果分区数越多，这部分缓存所需的内存占用也会更多。假设你有10000个分区，按照默认设置，这部分缓存需要占用约157MB的内存。而consumer端呢？我们抛开获取数据所需的内存不说，只说线程的开销。如果还是假设有10000个分区，同时consumer线程数要匹配分区数(大部分情况下是最佳的消费吞吐量配置)的话，那么在consumer client就要创建10000个线程，也需要创建大约10000个Socket去获取分区数据。这里面的线程切换的开销本身已经不容小觑了。
服务器端的开销也不小，如果阅读Kafka源码的话可以发现，服务器端的很多组件都在内存中维护了分区级别的缓存，比如controller，FetcherManager等，因此分区数越多，这种缓存的成本越久越大。
二、文件句柄的开销
每个分区在底层文件系统都有属于自己的一个目录。该目录下通常会有两个文件： base_offset.log和base_offset.index。Kafak的controller和ReplicaManager会为每个broker都保存这两个文件句柄(file handler)。很明显，如果分区数越多，所需要保持打开状态的文件句柄数也就越多，最终可能会突破你的ulimit -n的限制。
三、降低高可用性
Kafka通过副本(replica)机制来保证高可用。具体做法就是为每个分区保存若干个副本(replica_factor指定副本数)。每个副本保存在不同的broker上。期中的一个副本充当leader 副本，负责处理producer和consumer请求。其他副本充当follower角色，由Kafka controller负责保证与leader的同步。如果leader所在的broker挂掉了，contorller会检测到然后在zookeeper的帮助下重选出新的leader——这中间会有短暂的不可用时间窗口，虽然大部分情况下可能只是几毫秒级别。但如果你有10000个分区，10个broker，也就是说平均每个broker上有1000个分区。此时这个broker挂掉了，那么zookeeper和controller需要立即对这1000个分区进行leader选举。比起很少的分区leader选举而言，这必然要花更长的时间，并且通常不是线性累加的。如果这个broker还同时是controller情况就更糟了。
说了这么多“废话”，很多人肯定已经不耐烦了。那你说到底要怎么确定分区数呢？答案就是：视情况而定。基本上你还是需要通过一系列实验和测试来确定。当然测试的依据应该是吞吐量。虽然LinkedIn这篇文章做了Kafka的基准测试，但它的结果其实对你意义不大，因为不同的硬件、软件、负载情况测试出来的结果必然不一样。我经常碰到的问题类似于，官网说每秒能到10MB，为什么我的producer每秒才1MB？ —— 且不说硬件条件，最后发现他使用的消息体有1KB，而官网的基准测试是用100B测出来的，因此根本没有可比性。不过你依然可以遵循一定的步骤来尝试确定分区数：创建一个只有1个分区的topic，然后测试这个topic的producer吞吐量和consumer吞吐量。假设它们的值分别是Tp和Tc，单位可以是MB/s。然后假设总的目标吞吐量是Tt，那么分区数 = Tt / max(Tp, Tc)
Tp表示producer的吞吐量。测试producer通常是很容易的，因为它的逻辑非常简单，就是直接发送消息到Kafka就好了。Tc表示consumer的吞吐量。测试Tc通常与应用的关系更大， 因为Tc的值取决于你拿到消息之后执行什么操作，因此Tc的测试通常也要麻烦一些。
另外，Kafka并不能真正地做到线性扩展(其实任何系统都不能)，所以你在规划你的分区数的时候最好多规划一下，这样未来扩展时候也更加方便。
消息-分区的分配
默认情况下，Kafka根据传递消息的key来进行分区的分配，即hash(key) % numPartitions，如下图所示:
def partition(key: Any, numPartitions: Int): Int = {
Utils.abs(key.hashCode) % numPartitions
}

这就保证了相同key的消息一定会被路由到相同的分区。如果你没有指定key，那么Kafka是如何确定这条消息去往哪个分区的呢？
if(key == null) { // 如果没有指定key
val id = sendPartitionPerTopicCache.get(topic) // 先看看Kafka有没有缓存的现成的分区Id
id match {
case Some(partitionId) =>
partitionId // 如果有的话直接使用这个分区Id就好了
case None => // 如果没有的话，
val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined) //找出所有可用分区的leader所在的broker
if (availablePartitions.isEmpty)
throw new LeaderNotAvailableException("No leader for any partition in topic " + topic)
val index = Utils.abs(Random.nextInt) % availablePartitions.size // 从中随机挑一个
val partitionId = availablePartitions(index).partitionId
sendPartitionPerTopicCache.put(topic, partitionId) // 更新缓存以备下一次直接使用
partitionId
}
}

可以看出，Kafka几乎就是随机找一个分区发送无key的消息，然后把这个分区号加入到缓存中以备后面直接使用——当然了，Kafka本身也会清空该缓存（默认每10分钟或每次请求topic元数据时）
如何设定consumer线程数
我个人的观点，如果你的分区数是N，那么最好线程数也保持为N，这样通常能够达到最大的吞吐量。超过N的配置只是浪费系统资源，因为多出的线程不会被分配到任何分区。让我们来看看具体Kafka是如何分配的。
topic下的一个分区只能被同一个consumer group下的一个consumer线程来消费，但反之并不成立，即一个consumer线程可以消费多个分区的数据，比如Kafka提供的ConsoleConsumer，默认就只是一个线程来消费所有分区的数据。——其实ConsoleConsumer可以使用通配符的功能实现同时消费多个topic数据，但这和本文无关。
再讨论分配策略之前，先说说KafkaStream——它是consumer的关键类，提供了遍历方法用于consumer程序调用实现数据的消费。其底层维护了一个阻塞队列，所以在没有新消息到来时，consumer是处于阻塞状态的，表现出来的状态就是consumer程序一直在等待新消息的到来。——你当然可以配置成带超时的consumer，具体参看参数consumer.timeout.ms的用法。
下面说说 Kafka提供的两种分配策略： range和roundrobin，由参数partition.assignment.strategy指定，默认是range策略。本文只讨论range策略。所谓的range其实就是按照阶段平均分配。举个例子就明白了，假设你有10个分区，P0 ~ P9，consumer线程数是3， C0 ~ C2，那么每个线程都分配哪些分区呢？
C0 消费分区 0, 1, 2, 3
C1 消费分区 4, 5, 6
C2 消费分区 7, 8, 9
具体算法就是：
val nPartsPerConsumer = curPartitions.size / curConsumers.size // 每个consumer至少保证消费的分区数
val nConsumersWithExtraPart = curPartitions.size % curConsumers.size // 还剩下多少个分区需要单独分配给开头的线程们
...
for (consumerThreadId <- consumerThreadIdSet) { // 对于每一个consumer线程
val myConsumerPosition = curConsumers.indexOf(consumerThreadId) //算出该线程在所有线程中的位置，介于[0, n-1]
assert(myConsumerPosition >= 0)
// startPart 就是这个线程要消费的起始分区数
val startPart = nPartsPerConsumer * myConsumerPosition + myConsumerPosition.min(nConsumersWithExtraPart)
// nParts 就是这个线程总共要消费多少个分区
val nParts = nPartsPerConsumer + (if (myConsumerPosition + 1 > nConsumersWithExtraPart) 0 else 1)
...
}


针对于这个例子，nPartsPerConsumer就是10/3=3，nConsumersWithExtraPart为10%3=1，说明每个线程至少保证3个分区，还剩下1个分区需要单独分配给开头的若干个线程。这就是为什么C0消费4个分区，后面的2个线程每个消费3个分区，具体过程详见下面的Debug截图信息：
ctx.myTopicThreadIds



nPartsPerConsumer = 10 / 3 = 3
nConsumersWithExtraPart = 10 % 3 = 1


第一次：
myConsumerPosition = 1
startPart = 1 * 3 + min(1, 1) = 4 ---也就是从分区4开始读
nParts = 3 + (if (1 + 1 > 1) 0 else 1) = 3 读取3个分区， 即4,5,6
第二次：
myConsumerPosition = 0
startPart = 3 * 0 + min(1, 0) =0 --- 从分区0开始读
nParts = 3 + (if (0 + 1 > 1) 0 else 1) = 4 读取4个分区，即0,1,2,3
第三次：
myConsumerPosition = 2
startPart = 3 * 2 + min(2, 1) = 7 --- 从分区7开始读
nParts = 3 + if (2 + 1 > 1) 0 else 1) = 3 读取3个分区，即7, 8, 9
至此10个分区都已经分配完毕


说到这里，经常有个需求就是我想让某个consumer线程消费指定的分区而不消费其他的分区。坦率来说，目前Kafka并没有提供自定义分配策略。做到这点很难，但仔细想一想，也许我们期望Kafka做的事情太多了，毕竟它只是个消息引擎，在Kafka中加入消息消费的逻辑也许并不是Kafka该做的事情。

不消费问题
第一步：参看消费者的基本情况

查看mwbops系统，【Consumer监控】-->【对应的consumerId】

如果offset数字一直在动，说明一直在消费，说明不存在问题，return;
如果offset数字一直不动，看Owner是不是有值存在
如果Owner是空，说明消费端的程序已经跟Kafka断开连接，应该排查消费端是否正常，return;
如果Owner不为空，就是有上图上面的类似于 bennu_index_benuprdapp02-1444748505181-f558155a-0 的文字，继续看下面内容
第二步：查看消费端的程序代码
一般的消费代码是这样的

看看自己的消费代码里面，存不存在处理消息的时候出异常的情况
如果有，需要try-catch一下，其实不论有没有异常，都用try-catch包一下最好，如下面代码

return;
原因：如果在处理消息的时候有异常出现，又没有进行处理，那么while循环就会跳出，线程会结束，所以不会再去取消息，就是消费停止了。
第三步：查看消费端的配置
消费代码中一般以以下方式创建Consumer

消费端有一个配置，叫 fetch.message.max.bytes，默认是1M，此时如果有消息大于1M，会发生停止消费的情况。
此时，在配置中增加 props.put("fetch.message.max.bytes", "10 * 1024 * 1024"); 即可
return;
原因：目前Kafka集群配置的运行最大的消息大小是10M，如果客户端配置的运行接收的消息是1M，跟Kafka服务端配置的不一致，
则消息大于1M的情况下，消费端就无法消费，导致一直卡在这一条消息，现象就是消费停止。

kafka查看消费了多少条数据

情况是这样的，在我们系统中有多个Consumer的客户端（客户端个数是不确定的，因为在系统工作过程中有的业务节点会脱离，有些业务节点会增加进来），Producer也有多个。但是Producer发送的消息种类只有一种，所以topic只创建了一个， 消息量很大，所以使用了多个Consumer来处理。现在想实现如下的订阅/推送效果，多个Producer进行消息的推送，例如消息X1、X2、X3、X4、X5.。。。。。。然后由多个Consumer分别进行拉去，Consumer1拉取到：X1、X4、X7。。。Consumer2拉取到：X2、X5、X8.。。。。。。。。如此类推

kafkaspout消费过的数据怎么还消费

建议去看下这边帖子：http://blog.csdn.net/zollty/article/details/53958641


Kafka重复消费原因
底层根本原因：已经消费了数据，但是offset没提交。
原因1：强行kill线程，导致消费后的数据，offset没有提交。
原因2：设置offset为自动提交，关闭kafka时，如果在close之前，调用 consumer.unsubscribe() 则有可能部分offset没提交，下次重启会重复消费。例如：
try {
consumer.unsubscribe();
} catch (Exception e) {
}
try {
consumer.close();
} catch (Exception e) {
}
上面代码会导致部分offset没提交，下次启动时会重复消费。
原因3（重复消费最常见的原因）：消费后的数据，当offset还没有提交时，partition就断开连接。比如，通常会遇到消费的数据，处理很耗时，导致超过了Kafka的session timeout时间（0.10.x版本默认是30秒），那么就会re-blance重平衡，此时有一定几率offset没提交，会导致重平衡后重复消费。



Kafka Consumer丢失数据原因
猜测：设置offset为自动定时提交，当offset被自动定时提交时，数据还在内存中未处理，此时刚好把线程kill掉，那么offset已经提交，但是数据未处理，导致这部分内存中的数据丢失。



解决方案：记录offset和恢复offset的方案

理论上记录offset，下一个group consumer可以接着记录的offset位置继续消费。
offset记录方案：
每次消费时更新每个topic+partition位置的offset在内存中，
Map，key=topic+'-'+partition，value=offset
当调用关闭consumer线程时，把上面Map的offset数据记录到 文件中*（分布式集群可能要记录到Redis中）。

下一次启动consumer，需要读取上一次的offset信息，方法是 以当前的topic+partition为key，从上次的Map中去寻找offset。
然后使用consumer.seek()方法指定到上次的offset位置。

说明：
1、该方案针对单台服务器比较简单，直接把offset记录到本地文件中即可，但是对于多台服务器集群，offset也要记录到同一个地方，并且需要做去重处理。
如果线上程序是由多台服务器组成的集群，是否可以用一台服务器来支撑？应该可以，只是消费慢一点，没多大影响。

2、如何保证接着offset消费的数据正确性
为了确保consumer消费的数据一定是接着上一次consumer消费的数据，
consumer消费时，记录第一次取出的数据，将其offset和上次consumer最后消费的offset进行对比，如果相同则继续消费。如果不同，则停止消费，检查原因。

kafka获取数据的几种方式

一、基于Receiver的方式

这种方式使用Receiver来获取数据。Receiver是使用Kafka的高层次Consumer API来实现的。receiver从Kafka中获取的数据都是存储在Spark Executor的内存中的，然后Spark Streaming启动的job会去处理那些数据。

然而，在默认的配置下，这种方式可能会因为底层的失败而丢失数据。如果要启用高可靠机制，让数据零丢失，就必须启用Spark Streaming的预写日志机制（Write Ahead Log，WAL）。该机制会同步地将接收到的Kafka数据写入分布式文件系统（比如HDFS）上的预写日志中。所以，即使底层节点出现了失败，也可以使用预写日志中的数据进行恢复。

如何进行Kafka数据源连接

1、在maven添加依赖

org.apache.spark spark-streaming-kafka_2.10 1.4.1


2、scala代码

val kafkaStream = {val sparkStreamingConsumerGroup = "spark-streaming-consumer-group"val kafkaParams = Map("zookeeper.connect" -> "zookeeper1:2181","group.id" -> "spark-streaming-test","zookeeper.connection.timeout.ms" -> "1000")val inputTopic = "input-topic"val numPartitionsOfInputTopic = 5val streams = (1 to numPartitionsOfInputTopic) map { _ =>KafkaUtils.createStream(ssc, kafkaParams, Map(inputTopic -> 1), StorageLevel.MEMORY_ONLY_SER).map(_._2)}val unifiedStream = ssc.union(streams)val sparkProcessingParallelism = 1 // You'd probably pick a higher value than 1 in production.unifiedStream.repartition(sparkProcessingParallelism)}


需要注意的要点

1、Kafka中的topic的partition，与Spark中的RDD的partition是没有关系的。所以，在KafkaUtils.createStream()中，提高partition的数量，只会增加一个Receiver中，读取partition的线程的数量。不会增加Spark处理数据的并行度。

2、可以创建多个Kafka输入DStream，使用不同的consumer group和topic，来通过多个receiver并行接收数据。

3、如果基于容错的文件系统，比如HDFS，启用了预写日志机制，接收到的数据都会被复制一份到预写日志中。因此，在KafkaUtils.createStream()中，设置的持久化级别是StorageLevel.MEMORY_AND_DISK_SER。

二、基于Direct的方式

这种新的不基于Receiver的直接方式，是在Spark 1.3中引入的，从而能够确保更加健壮的机制。替代掉使用Receiver来接收数据后，这种方式会周期性地查询Kafka，来获得每个topic+partition的最新的offset，从而定义每个batch的offset的范围。当处理数据的job启动时，就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。

这种方式有如下优点：

1、简化并行读取：如果要读取多个partition，不需要创建多个输入DStream然后对它们进行union操作。Spark会创建跟Kafka partition一样多的RDD partition，并且会并行从Kafka中读取数据。所以在Kafka partition和RDD partition之间，有一个一对一的映射关系。

2、高性能：如果要保证零数据丢失，在基于receiver的方式中，需要开启WAL机制。这种方式其实效率低下，因为数据实际上被复制了两份，Kafka自己本身就有高可靠的机制，会对数据复制一份，而这里又会复制一份到WAL中。而基于direct的方式，不依赖Receiver，不需要开启WAL机制，只要Kafka中作了数据的复制，那么就可以通过Kafka的副本进行恢复。

3、一次且仅一次的事务机制：

基于receiver的方式，是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的传统方式。这种方式配合着WAL机制可以保证数据零丢失的高可靠性，但是却无法保证数据被处理一次且仅一次，可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。

基于direct的方式，使用kafka的简单api，Spark Streaming自己就负责追踪消费的offset，并保存在checkpoint中。Spark自己一定是同步的，因此可以保证数据是消费一次且仅消费一次。

scala连接代码

val topics = Set("teststreaming")val brokers = "bdc46.hexun.com:9092,bdc53.hexun.com:9092,bdc54.hexun.com:9092" val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers, "serializer.class" -> "kafka.serializer.StringEncoder")// Create a direct stream val kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)val events = kafkaStream.flatMap(line => {Some(line.toString())})


三、总结：两种方式在生产中都有广泛的应用，新api的Direct应该是以后的首选方式。

如何确定kafka生产者确实发送了数据

Kafka的生产者和消费者都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此分区实际上是调优Kafka并行度的最小单元。对于producer而言，它实际上是用多个线程并发地向不同分区所在的broker发起Socket连接同时给这些分区发送消息；而consumer呢，同一个消费组内的所有consumer线程都被指定topic的某一个分区进行消费(具体如何确定consumer线程数目我们后面会详细说明)。所以说，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。

怎么查看kafka集群中所有的broker节点

　　在Zookeeper的官 网上有这么一句话：ZooKeeper is a centralized service for maintaining configuration information, naming, providing distributed synchronization, and providing group services.
　　这大概描述了Zookeeper主要可以干哪些事情：配置管理，名字服务，提供分布式同步以及集群管理。那这些服务又到底是什么呢？我们为什么需要这样的服务？我们又为什么要使用Zookeeper来实现呢，使用Zookeeper有什么优势？接下来我会挨个介绍这些到底是什么，以及有哪些开源系统中使用了。
　　配置管理
　　在我们的应用中除了代码外，还有一些就是各种配置。比如数据库连接等。一般我们都是使用配置文件的方式，在代码中引入这些配置文件。但是当我们只有一种配置，只有一台服务器，并且不经常修改的时候，使用配置文件是一个很好的做法，但是如果我们配置非常多，有很多服务器都需要这个配置，而且还可能是动态的话使用配置文件就不是个好主意了。这个时候往往需要寻找一种集中管理配置的方法，我们在这个集中的地方修改了配置，所有对这个配置感兴趣的都可以获得变更。比如我们可以把配置放在数据库里，然后所有需要配置的服务都去这个数据库读取配置。但是，因为很多服务的正常运行都非常依赖这个配置，所以需要这个集中提供配置服务的服务具备很高的可靠性。一般我们可以用一个集群来提供这个配置服务，但是用集群提升可靠性，那如何保证配置在集群中的一致性呢？ 这个时候就需要使用一种实现了一致性协议的服务了。Zookeeper就是这种服务，它使用Zab这种一致性协议来提供一致性。现在有很多开源项目使用Zookeeper来维护配置，比如在HBase中，客户端就是连接一个Zookeeper，获得必要的HBase集群的配置信息，然后才可以进一步操作。还有在开源的消息队列Kafka中，也使用Zookeeper来维护broker的信息。在Alibaba开源的SOA框架Dubbo中也广泛的使用Zookeeper管理一些配置来实现服务治理。
　　名字服务
　　名字服务这个就很好理解了。比如为了通过网络访问一个系统，我们得知道对方的IP地址，但是IP地址对人非常不友好，这个时候我们就需要使用域名来访问。但是计算机是不能是别域名的。怎么办呢？如果我们每台机器里都备有一份域名到IP地址的映射，这个倒是能解决一部分问题，但是如果域名对应的IP发生变化了又该怎么办呢？于是我们有了DNS这个东西。我们只需要访问一个大家熟知的(known)的点，它就会告诉你这个域名对应的IP是什么。在我们的应用中也会存在很多这类问题，特别是在我们的服务特别多的时候，如果我们在本地保存服务的地址的时候将非常不方便，但是如果我们只需要访问一个大家都熟知的访问点，这里提供统一的入口，那么维护起来将方便得多了。
　　分布式锁
　　其实在第一篇文章中已经介绍了Zookeeper是一个分布式协调服务。这样我们就可以利用Zookeeper来协调多个分布式进程之间的活动。比如在一个分布式环境中，为了提高可靠性，我们的集群的每台服务器上都部署着同样的服务。但是，一件事情如果集群中的每个服务器都进行的话，那相互之间就要协调，编程起来将非常复杂。而如果我们只让一个服务进行操作，那又存在单点。通常还有一种做法就是使用分布式锁，在某个时刻只让一个服务去干活，当这台服务出问题的时候锁释放，立即fail over到另外的服务。这在很多分布式系统中都是这么做，这种设计有一个更好听的名字叫Leader Election(leader选举)。比如HBase的Master就是采用这种机制。但要注意的是分布式锁跟同一个进程的锁还是有区别的，所以使用的时候要比同一个进程里的锁更谨慎的使用。
　　集群管理
　　在分布式的集群中，经常会由于各种原因，比如硬件故障，软件故障，网络问题，有些节点会进进出出。有新的节点加入进来，也有老的节点退出集群。这个时候，集群中其他机器需要感知到这种变化，然后根据这种变化做出对应的决策。比如我们是一个分布式存储系统，有一个中央控制节点负责存储的分配，当有新的存储进来的时候我们要根据现在集群目前的状态来分配存储节点。这个时候我们就需要动态感知到集群目前的状态。还有，比如一个分布式的SOA架构中，服务是一个集群提供的，当消费者访问某个服务时，就需要采用某种机制发现现在有哪些节点可以提供该服务(这也称之为服务发现，比如Alibaba开源的SOA框架Dubbo就采用了Zookeeper作为服务发现的底层机制)。还有开源的Kafka队列就采用了Zookeeper作为Cosnumer的上下线管理。
　　后记
　　在这篇文章中，列出了一些Zookeeper可以提供的服务，并给出了一些开源系统里面的实例。后面我们从Zookeeper的安装配置开始，并用示例进一步介绍Zookeeper如何使用。
　　(转载）

kafka查看消费了多少条数据

如何查看目前的消费者是否已经读到最新的数据：


kafka-run-class.sh kafka.tools.ConsumerOffsetChecker
#kafka查看topic各个分区的消息的信息
kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group ** --topic *** --zookeeper *:2181,*:2181,*:2181/kafka
--zookeeper 那里是指kafka在zk中的path,即使zk有多个机器，如果在其中一台上执行此命令，显示连接不上，只写那台机器的地址端口+kafka的path即可
指定自己的分组 自己消费的topic会显示kafka总共有多少数据，以及已经被消费了多少条
结果：
GROUP TOPIC PID OFFSET LOGSIZE LAG
消费者组 话题id 分区id 当前已消费的条数 总条数 未消费的条数



注意：以kafkaspout类作为消费者去读kafka数据，相当于直接从kafka server上取文件，没有消费者组的概念
每次读的数据存在自己zk的offet中
所以不能通过上述命令查看

linux 怎样查看kafka的某topic数据？

查看topic分布情况kafka-list-topic.sh： bin/kafka-list-topic.sh-zookeeper 192.168.197.170:2181,192.168.197.171:2181 （列出所有topic的分区情况） bin/kafka-list-topic.sh-zookeeper 192.168.197.170:2181,192.168.197.171:2181-topic test （查看test的分区情况） 创建topickafka-create-topic.sh： bin/kafka-create-topic.sh-replica 2-partition 8-topic test-zookeeper 192.168.197.170:2181,192.168.197.171:2181 创建名为test的topic，8个分区分别存放数据，数据备份总共2份。 参考资料 kafka 创建topic,查看topic - CSDN博客.blog.csdn.[引用时间2017-12-29]

kafka消费者java版本读取不到消息怎么办

3. 启动服务
3.1 启动zookeeper
启动zk有两种方式，第一种是使用kafka自己带的一个zk。
bin/zookeeper-server-start.sh config/zookeeper.properties&
另一种是使用其它的zookeeper，可以位于本机也可以位于其它地址。这种情况需要修改config下面的sercer.properties里面的zookeeper地址
。例如zookeeper.connect=10.202.4.179:2181
3.2 启动 kafka
bin/kafka-server-start.sh config/server.properties
4.创建topic
bin/kafka-topics.sh --create --zookeeper 10.202.4.179:2181 --replication-factor 1 --partitions 1 --topic test
创建一个名为test的topic，只有一个副本，一个分区。
通过list命令查看刚刚创建的topic
bin/kafka-topics.sh -list -zookeeper 10.202.4.179:2181
5.启动producer并发送消息启动producer
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
启动之后就可以发送消息了
比如
test
hello boy
按Ctrl+C退出发送消息
6.启动consumer
bin/kafka-console-consumer.sh --zookeeper 10.202.4.179:2181 --topic test --from-beginning
启动consumer之后就可以在console中看到producer发送的消息了
可以开启两个终端，一个发送消息，一个接受消息。
如果这样都不行的话，查看zookeeper进程和kafka的topic，一步步排查原因吧。