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，一步步排查原因吧。

数据丢失是一件非常严重的事情事，针对数据丢失的问题我们需要有明确的思路来确定问题所在，针对这段时间的总结，我个人面对kafka 数据丢失问题的解决思路如下：

1、是否真正的存在数据丢失问题，比如有很多时候可能是其他同事 *** 作了测试环境，所以首先确保数据没有第三方干扰。

2、理清你的业务流程，数据流向，数据到底是在什么地方丢失的数据，在kafka 之前的环节或者kafka之后的流程丢失？比如kafka的数据是由flume提供的，也许是flume丢失了数据，kafka 自然就没有这一部分数据。

3、如何发现有数据丢失，又是如何验证的。从业务角度考虑，例如：教育行业，每年高考后数据量巨大，但是却反常的比高考前还少，或者源端数据量和目的端数据量不符

4、 定位数据是否在kafka之前就已经丢失还事消费端丢失数据的

kafka支持数据的重新回放功能(换个消费group)，清空目的端所有数据，重新消费。如果是在消费端丢失数据，那么多次消费结果完全一模一样的几率很低。如果是在写入端丢失数据，那么每次结果应该完全一样(在写入端没有问题的前提下)。

5、kafka环节丢失数据，常见的kafka环节丢失数据的原因有：

如果auto.commit.enable=true，当consumer fetch了一些数据但还没有完全处理掉的时候，刚好到commit interval出发了提交offset *** 作，接着consumer crash掉了。这时已经fetch的数据还没有处理完成但已经被commit掉，因此没有机会再次被处理，数据丢失。网络负载很高或者磁盘很忙写入失败的情况下，没有自动重试重发消息。没有做限速处理，超出了网络带宽限速。kafka一定要配置上消息重试的机制，并且重试的时间间隔一定要长一些，默认1秒钟并不符合生产环境（网络中断时间有可能超过1秒）。如果磁盘坏了，会丢失已经落盘的数据

单批数据的长度超过限制会丢失数据，报kafka.common.MessageSizeTooLargeException异常解决：

6、partition leader在未完成副本数follows的备份时就宕机的情况，即使选举出了新的leader但是已经push的数据因为未备份就丢失了！kafka是多副本的，当你配置了同步复制之后。多个副本的数据都在PageCache里面，出现多个副本同时挂掉的概率比1个副本挂掉的概率就很小了。（官方推荐是通过副本来保证数据的完整性的）

7、kafka的数据一开始就是存储在PageCache上的，定期flush到磁盘上的，也就是说，不是每个消息都被存储在磁盘了，如果出现断电或者机器故障等，PageCache上的数据就丢失了。可以通过log.flush.interval.messages和log.flush.interval.ms来配置flush间隔，interval大丢的数据多些，小会影响性能但在0.8版本，可以通过replica机制保证数据不丢，代价就是需要更多资源，尤其是磁盘资源，kafka当前支持GZip和Snappy压缩，来缓解这个问题 是否使用replica取决于在可靠性和资源代价之间的balance。

同时kafka也提供了相关的配置参数，来让你在性能与可靠性之间权衡（一般默认）：

当达到下面的消息数量时，会将数据flush到日志文件中。默认10000

当达到下面的时间(ms)时，执行一次强制的flush *** 作。interval.ms和interval.messages无论哪个达到，都会flush。默认3000ms

检查是否需要将日志flush的时间间隔

high-level版本已经封装了对partition和offset的管理，默认是会定期自动commit offset，这样可能会丢数据的low-level版本自己管理spout线程和partition之间的对应关系和每个partition上的已消费的offset(定期写到zk)并且只有当这个offset被ack后，即成功处理后，才会被更新到zk，所以基本是可以保证数据不丢的即使spout线程crash(崩溃)，重启后还是可以从zk中读到对应的offset

不能让内存的缓冲池太满，如果满了内存溢出，也就是说数据写入过快，kafka的缓冲池数据落盘速度太慢，这时肯定会造成数据丢失。尽量保证生产者端数据一直处于线程阻塞状态，这样一边写内存一边落盘。异步写入的话还可以设置类似flume回滚类型的batch数，即按照累计的消息数量，累计的时间间隔，累计的数据大小设置batch大小。

不过异步写入丢失数据的情况还是难以控制还是得稳定整体集群架构的运行，特别是zookeeper，当然正对异步数据丢失的情况尽量保证broker端的稳定运作吧

kafka不像hadoop更致力于处理大量级数据，kafka的消息队列更擅长于处理小数据。针对具体业务而言，若是源源不断的push大量的数据（eg：网络爬虫），可以考虑消息压缩。但是这也一定程度上对CPU造成了压力,还是得结合业务数据进行测试选择

topic设置多分区，分区自适应所在机器，为了让各分区均匀分布在所在的broker中，分区数要大于broker数。分区是kafka进行并行读写的单位，是提升kafka速度的关键。

关闭自动更新offset，等到数据被处理后再手动跟新offset。

在消费前做验证前拿取的数据是否是接着上回消费的数据，不正确则return先行处理排错。

一般来说zookeeper只要稳定的情况下记录的offset是没有问题，除非是多个consumer group 同时消费一个分区的数据，其中一个先提交了，另一个就丢失了。

kafka的数据一开始就是存储在PageCache上的，定期flush到磁盘上的，也就是说，不是每个消息都被存储在磁盘了，如果出现断电或者机器故障等，PageCache上的数据就丢失了。这个是总结出的到目前为止没有发生丢失数据的情况

强行kill线程，导致消费后的数据，offset没有提交，partition就断开连接。比如，通常会遇到消费的数据，处理很耗时，导致超过了Kafka的session timeout时间（0.10.x版本默认是30秒），那么就会re-blance重平衡，此时有一定几率offset没提交，会导致重平衡后重复消费。

如果在close之前调用了consumer.unsubscribe()则有可能部分offset没提交，下次重启会重复消费。

kafka数据重复 kafka设计的时候是设计了(at-least once)至少一次的逻辑，这样就决定了数据可能是重复的，kafka采用基于时间的SLA(服务水平保证)，消息保存一定时间（通常为7天）后会被删除。

kafka的数据重复一般情况下应该在消费者端,这时log.cleanup.policy = delete使用定期删除机制。

publicstaticvoidconsumer(){Propertiesprops=newProperties()props.put("zk.connect","hadoop-2:2181")props.put("zk.connectiontimeout.ms","1000000")props.put("groupid","fans_group")//CreatetheconnectiontotheclusterConsumerConfigconsumerConfig=newConsumerConfig(props)ConsumerConnectorconsumerConnector=Consumer.createJavaConsumerConnector(consumerConfig)Mapmap=newHashMap()map.put("fans",1)//create4partitionsofthestreamfortopic“test”,toallow4threadstoconsumeMap>>topicMessageStreams=consumerConnector.createMessageStreams(map)List>streams=topicMessageStreams.get("fans")//createlistof4threadstoconsumefromeachofthepartitionsExecutorServiceexecutor=Executors.newFixedThreadPool(1)longstartTime=System.currentTimeMillis()//consumethemessagesinthethreadsfor(finalKafkaStreamstream:streams){executor.submit(newRunnable(){publicvoidrun(){ConsumerIteratorit=stream.iterator()while(it.hasNext()){log.debug(byteBufferToString(it.next().message().payload()))}}})log.debug("usetime="+(System.currentTimeMillis()-startTime))}}

---