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

目录
  • kafka消费者kafka-console-consumer接收不到数据
    • 问题
    • 解决
  • 关于kafka-console-consumer.sh消费者的一些思考
    • (人物设定初步了解kafka的我)
    • kafka-console-consumer.sh相关知识拓展
  • 总结

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。
参数 值类型 说明 有效值
--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以允许故障转移

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持我们。

赞 (0)

相关推荐

  • kafka消费不到数据的排查过程

    目录 kafka消费不到数据的排查 流程大概说就是 总结 kafka消费不到数据的排查 集群上新安装并启动了3个kafka Broker,代码打包上传至集群,运行后发现一直消费不到数据, 本地idea中debug后发现,程序一直阻塞在如下程序中,陷入了死循环. /** * Block until the coordinator for this group is known and is ready to receive requests. * 等待直到我们和服务端的GroupCoordinat

  • Java Flink与kafka实现实时告警功能过程

    目录 引出问题 demo设计 环境搭建 flink程序代码 项目演示 告警系统架构 引出问题 项目使用告警系统的逻辑是将实时数据保存到本地数据库再使用定时任务做判断,然后产生告警数据.这种方式存在告警的延时实在是太高了.数据从产生到保存,从保存到判断都会存在时间差,按照保存数据定时5分钟一次,定时任务5分钟一次.最高会产生10分钟的误差,这种告警就没什么意义了. demo设计 为了简单的还原业务场景,做了简单的demo假设 实现一个对于学生成绩评价的实时处理程序 数学成绩,基准范围是90-140

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

    目录 kafka消费者kafka-console-consumer接收不到数据 问题 解决 关于kafka-console-consumer.sh消费者的一些思考 (人物设定初步了解kafka的我) kafka-console-consumer.sh相关知识拓展 总结 kafka消费者kafka-console-consumer接收不到数据 发送端 接收端 问题 采用内置的zookeeper,发送端发送数据,接收端能够接收数据 但是采用外置的zookeeper,发送端发送数据,接收端一直接收不到

  • 关于Kafka消费者订阅方式

    关于Kafka消费者订阅方式

    目录 Kafka消费者订阅方式 1.指定主题消费 2.指定分区消费 3.取消订阅 Kafka概述 定义 消息队列 1.传统消息队列的应用场景 2.消息队列的两种模式 Kafka 基础架构 Kafka消费者订阅方式 Kafka为消费者提供了三种类型的订阅消费方式:订阅主题集合.正则表达式订阅主题.订阅指定主题的分区集合.三种方式只能使用其中一种. 1.指定主题消费 一个消费者可以使用KafkaConsumer提供的subscribe()方法订阅一个或多个主题,订阅主题集合和正则表达式订阅主题都使用

  • axios发送post请求springMVC接收不到参数的解决方法

    axios发送post请求时,出现了参数后台接收不到的情况,分析了下请求,发现是请求头content-type不对,是application/json,正常应该是application/x-www-form-urlencoded. 解决方法有以下三种: 1.设置axios的默认请求头 //设置全局的 axios.defaults.headers.post['Content-Type'] = 'application/x-www-form-urlencoded'; var instance = a

  • vue-socket.io接收不到数据问题的解决方法

    最近公司的一个vue项目用到了vue-socket.io来处理socket数据传输,之前用过socket.io-client,现在知道vue-socket.io是基于socket.io-client的一层封装,将socket挂于全局从而更方便的书写. 于是把代码拉取下来运行: 什么鬼,同样的代码为什么我的就接收不到数据,自己新建一个测试一下吧! 先用express和socket.io搭个小socket服务器: let express = require('express'); let app =

  • jquery的ajax异步请求接收返回json数据实例

    jquery的ajax异步请求接收返回json数据方法设置简单,一个是服务器处理程序是返回json数据,另一种就是ajax发送设置的datatype设置为jsonp格式数据或json格式都可以. 代码示例如下: 复制代码 代码如下: $('#send').click(function () {     $.ajax({         type : "GET",         url : "a.php",         dataType : "json

  • Ajax发送和接收二进制字节流数据的方法

    HTML5 Ajax 2.0标准中,增强了Ajax的许多功能,包括发送FormData数据,上传数据进度条等诸多功能.但实际上,Ajax可以字节发送二进制数据. 发送二进制数据 var oReq = new XMLHttpRequest(); oReq.open("POST", url, true); oReq.onload = function (oEvent) { // Uploaded. }; var blob = new Blob(['abc123'], {type: 'tex

  • jsp传参 servlet接收中文乱码问题的解决方法

    在公司实习了8个月,一直都是做android和h5的,但是发现做程序连一点服务都不会该怎么办,所以最近开始学起了java,不知道是不是因为框架学多了,现在看起springmvc框架比以前看起来简单太多了,这里我是准备从hibernate开始学习,毕竟数据是根本嘛,首先我用的是hibernate+servlet,但是在jsp页面传参到servlet的时候中文一直乱码,我尝试了好多方法,最后还是解决了. 第一,首先看清项目的编码,jsp页面的编码 第二,修改tomcat 下面的server.xml文

  • java使用Socket类接收和发送数据

    网络应用分为客户端和服务端两部分,而Socket类是负责处理客户端通信的Java类.通过这个类可以连接到指定IP或域名的服务器上,并且可以和服务器互相发送和接受数据.在本文及后面的数篇文章中将详细讨论Socket类的使用,内容包括Socket类基础.各式各样的连接方式.get和set方法.连接过程中的超时以及关闭网络连接等. 在本文中,我们将讨论使用Socket类的基本步骤和方法.一般网络客户端程序在连接服务程序时要进行以下三步操作. 连接服务器 发送和接收数据 关闭网络连接 一.连接服务器 在

  • 浅谈springMVC接收前端json数据的总结

    对于json对象类型(即JsonObject)的数据,springMVC主要有以下几种方式接收: 1.通过Map接收 @RequestMapping(value = "/getAllStudio" ) public void getAllStudio(@RequestBody Map<String, Integer> map ) { JSONObject json = new JSONObject(); Integer page = map.get("page&q

  • .net webapi接收xml格式数据的3种情况小结

    前言 ASP.NET Web API是​​一个框架,可以很容易构建达成了广泛的HTTP服务客户端,包括浏览器和移动设备.是构建RESTful应用程序的理想平台的.NET框架. 本文主要给大家介绍了关于.net webapi接收xml格式数据的相关内容,下面话不多说了,来一起看看详细的介绍吧 webapi 接收 xml 的三种方法 前段时间接到一个任务写一个小接口,要接收java端返回过来的短信xml数据. 刚拿到项目,我的第一想法是对方会以什么形式发送xml格式的数据给我呢,设想三种情况. 我一

随机推荐