java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > kafka消费者kafka-console-consumer接收不到数据

kafka消费者kafka-console-consumer接收不到数据的解决

作者:k55

这篇文章主要介绍了kafka消费者kafka-console-consumer接收不到数据的问题及解决方案,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教

kafka消费者kafka-console-consumer接收不到数据

发送端

接收端

问题

采用内置的zookeeper,发送端发送数据,接收端能够接收数据

但是采用外置的zookeeper,发送端发送数据,接收端一直接收不到数据

解决

先判断主题是否一致,如果一致就在关闭kafka

./kafka-server-stop.sh ../config/server.properties

修改一下配置,确保这些配置已加上,不要用localhost,在listeners的ip地址和端口号要和消费者,生产者的的地址端口号一直

vim ../config/server.propertiesst3

最后把log.dirs后面的文件删除或者重新换个地址

rm -rf /tmp/kafka

重新在前台启动kafka,注意查看打印在桌面的日志,有无报错信息

./kafka-server-start.sh ../config/server.properties

如果没有报错信息,启动正常,那么就可以在后台启动了

./kafka-server-start.sh -daemon ../config/server.properties

创建生产者

./kafka-console-producer.sh --broker-list 172.16.193.175:9092 --topic test3

创建消费者

./kafka-console-consumer.sh --bootstrap-server 172.16.193.175:9092 --topic test3 --from-beginning

关于kafka-console-consumer.sh消费者的一些思考

(人物设定初步了解kafka的我)

我司现在有三台kafka服务器作为一个集群

需求是我写了一个监听器去监听活动失败的情况,如果活动失败则调用一个统计接口 做数据统计

我需要从失败事件的随路数据中取一些数据,做一些判断.

现在我想从集群中看一下失败事件中的随路数据是否完整正确

于是,我xshell连接上了三台服务器并且运行以下命令

./kafka-console-consumer.sh --bootstrap-server broker1IP:9092 --topic topicname
 
./kafka-console-consumer.sh --bootstrap-server broker2IP:9092 --topic topicname
 
./kafka-console-consumer.sh --bootstrap-server broker3IP:9092 --topic topicname

发现只要发送一个事件三个服务器都可以收到事件中的消息

怪了,为什么三台都会显示.

我第一反应是:这是否是传说中的leader和follower 同步策略

我问了一下我的leader ,

leader:.....,你知道你这个命令是什么意思吗?

这个命令就是相当于创建了一个消费者去消费了队列中的消息!

你这个3个服务器相当于启动了3个消费者去消费了,同一个消息三次!

我:不对啊,同一个消息不能被消费三次啊!?

leader:........,你知道什么是消费者组吗?你这相当于三个消费者组 不信你看看

./kafka-console-consumer.sh -help
 
...
 
--group <String: consumer group id>      The consumer group id of the consumer. 
 
...

看到了么?这里可以指定消费者组,你不指定他就默认是一个新的消费者组

我: 牛啊!

kafka-console-consumer.sh相关知识拓展

kafka-console-consumer.sh脚本是一个简易的消费者控制台。该 shell 脚本的功能通过调用 kafka.tools 包下的 ConsoleConsumer 类,并将提供的命令行参数全部传给该类实现。

./kafka-console-consumer.sh --bootstrap-server node:9092 --topic topicName
 
//表示从 latest 位移位置开始消费该主题的所有分区消息,即仅消费正在写入的消息。
  
 
bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --from-beginning --topic topicName
 
// 表示从指定主题中有效的起始位移位置开始消费所有分区的消息。
 
 
bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --property print.key=true --topic topicName
 
// 消费出的消息结果将打印出消息体的 key 和 value。
参数值类型说明有效值
--topicstring被消费的topic
--whiteliststring正则表达式,指定要包含以供使用的主题的白名单
--partitioninteger指定分区
除非指定’–offset’,否则从分区结束(latest)开始消费
--offsetstring执行消费的起始offset位置
默认值:latest
--from-beginning从存在的最早消息开始,而不是从最新消息开始
--max-messagesinteger消费的最大数据量,若不指定,则持续消费下去
--timeout-msinteger在指定时间间隔内没有消息可用时退出
--bootstrap-serverstring必需(除非使用旧版本的消费者),要连接的服务器
--key-deserializerstring
--value-deserializerstring
--groupstring指定消费者所属组的ID
--zookeeperstring必需(仅当使用旧的使用者时)连接zookeeper的字符串。
可以给出多个URL以允许故障转移

总结

以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。

您可能感兴趣的文章:
阅读全文