这篇“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服务器作为一个集群
需求是我写了一个监听器去监听活动失败的情况,如果活动失败则调用一个统计接口 做数据统计
我需要从失败事件的随路数据中取一些数据,做一些判断.
现在我想从集群中看一下失败事件中的随路数据是否完整正确
于是,我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脚本是一个简易的消费者控制台。该 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。
| 参数 | 值类型 | 说明 | 有效值 |
|---|---|---|---|
| --topic | string | 被消费的topic | |
| --whitelist | string | 正则表达式，指定要包含以供使用的主题的白名单 | |
| --partition | integer | 指定分区 除非指定’–offset’，否则从分区结束(latest)开始消费 | |
| --offset | string | 执行消费的起始offset位置 默认值:latest | |
| --from-beginning | 从存在的最早消息开始，而不是从最新消息开始 | ||
| --max-messages | integer | 消费的最大数据量，若不指定，则持续消费下去 | |
| --timeout-ms | integer | 在指定时间间隔内没有消息可用时退出 | |
| --bootstrap-server | string | 必需(除非使用旧版本的消费者)，要连接的服务器 | |
| --key-deserializer | string | ||
| --value-deserializer | string | ||
| --group | string | 指定消费者所属组的ID | |
| --zookeeper | string | 必需(仅当使用旧的使用者时)连接zookeeper的字符串。 可以给出多个URL以允许故障转移 |
以上就是关于“kafka消费者kafka-console-consumer接收不到数据如何解决”这篇文章的内容，相信大家都有了一定的了解，希望小编分享的内容对大家有帮助，若想了解更多相关的知识内容，请关注亿速云行业资讯频道。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
