如果broker的地址不正确，KafkConsumer中没有消息，如何设置topic的超时？
在Kafka中，如果broker的地址不正确，KafkaConsumer将无法连接到正确的broker，因此无法接收到消息。在这种情况下，设置topic的超时时间是无效的，因为KafkaConsumer无法与broker建立连接。
要解决这个问题，首先需要确保broker的地址正确。可以通过检查KafkaConsumer的配置参数来确认broker地址是否正确设置。确保配置参数"bootstrap.servers"中指定了正确的broker地址。
如果broker地址正确，但KafkaConsumer仍然无法接收到消息，可能是由于其他原因导致的。可以尝试以下方法来设置topic的超时时间：
- 使用KafkaConsumer的"poll"方法设置超时时间：可以在调用"poll"方法时传入一个超时时间参数，指定等待消息的最长时间。例如，可以使用"consumer.poll(Duration.ofMillis(timeout))"来设置超时时间为timeout毫秒。
- 使用KafkaConsumer的"subscribe"方法设置超时时间：可以在调用"subscribe"方法时传入一个超时时间参数，指定等待分区分配的最长时间。例如，可以使用"consumer.subscribe(topics, new ConsumerRebalanceListener() {...}, Duration.ofMillis(timeout))"来设置超时时间为timeout毫秒。
- 使用KafkaConsumer的"poll"方法和"ConsumerRecords"的isEmpty方法结合使用：可以在调用"poll"方法后，使用"ConsumerRecords"的isEmpty方法判断是否接收到了消息。如果isEmpty返回true，表示没有接收到消息，可以根据需要进行超时处理。
需要注意的是，设置超时时间只是一种处理方式，具体的处理方法还取决于业务需求和实际情况。可以根据具体情况选择适合的处理方式。
关于腾讯云相关产品和产品介绍链接地址，可以参考腾讯云官方文档或咨询腾讯云的技术支持团队获取更详细的信息。
相关·内容
通过这种方法，我能够从特定的Kafka主题中抓取消息。formattedDate).build(); } return messages;
} 当我输入了不正确的代理地址时我如何在消费者上设置超时，使它不会每次等待80秒，而是将时间减少到5秒？ 我的意思是像这样的东西。value.deserialize
浏览 108提问于2021-10-09得票数 1
7回答
我使用的是0.9.0.0版本的Kafka，我想在不使用管理脚本kafka-console-consumer.sh的情况下统计主题中的消息数量。我已经尝试了answer 中的所有命令，但没有一个命令产生结果。有人能帮我吗？
浏览 44提问于2017-01-22得票数 28
我在两个不同的设备上有MQTT代理(Mosquitto)。他们正在工作。我在他们之间架起了桥梁。我可以使用pub_mosquitto将数据从一个发布到另一个。我想从像这样的internet broker(broker.hivemq.com)中获得任何主题的数据
我可以从该网站经纪人的任何主题中获取数据，但我不能同时将采取的数据发布到另一个我的经纪人。我想使用我的代理pub/sub。我的代理
浏览 1提问于2018-08-29得票数 1
topic test这是一条消息2016-08-24 18:15:27,441向topic test发送消息时出错: null，value: 2字节错误:包含1条记录的(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.TimeoutException:批次在向broker请求测试元数据时超时过期-0
为什么无法将消息<
浏览 13提问于2016-08-24得票数 0
来自我的kafka生产者的一些消息超过了默认的1MB，并且被代理拒绝。
我无法访问代理和消费者来进行更改。因此，我只需要在生产者级别进行更改。我正在考虑在我的producer中使用batch.size到5MB的snappy压缩。请让我知道这是否是正确的选择或任何其他方式可用于我的用例。谢谢
浏览 0提问于2020-07-23得票数 0
我有两个应用程序，第一个应用程序启动一个ActiveMQ broker ( )。import org.apache.activemq.broker.BrokerService;
import org.springframework.beans.factory.annotation.Autowiredret.addConnector("tcp:
浏览 0提问于2018-04-09得票数 2
回答已采纳
1回答
查询:将路由器接口作为kafka集群的生产者。问题2. kafka消费者可以在不指定主题的情况下拉取数据吗？我刚刚检查了Kafka broker配置是否有"auto.create.topics.enable“字段。如果我设置
浏览 1提问于2014-06-18得票数 0
我的问题是，如果我们必须在生产者-控制台文件中只提供一个IP地址，如下所示或者所有的三个ip地址
./kafka-console-producer.sh --broker-list 192.168.7.110:9092,192.168.5.110:9092,192.168.3.111:90
浏览 0提问于2019-02-07得票数 3
回答已采纳
1回答
我是Kafa和数据摄取的新手。我知道Kafka是容错的，因为它将数据冗余地保存在多个节点上。然而，我不明白的是，我们如何在源/生产者端实现容错。例如，如果我使用netcat作为资源，如下面的示例所示。如果</em
浏览 6提问于2020-05-21得票数 2
我从我的python脚本连接到MySQL，从mqtt broker订阅一个主题，每当我收到数据时，我就会存储这些数据。","topic")port = 8888
client = mqtt.Client(clean_session因此，使用此代码可以在mqtt接收到任何消息时保存数据。但是有时候mqtt超过8小时<em
浏览 3提问于2019-10-17得票数 1
我今天遇到了这个数据问题，为了解决这个问题，我必须重新计算过去3个月的所有数据。但是，在Kafka中，当我运行以下命令时：它遇到一个错误requested offset is not within the range of offsets maintained by the server
--from-beginning正在尝试从数据
浏览 0提问于2018-08-16得票数 1
1回答
我使用kafka作为微服务体系结构的消息总线，因此多个服务侦听一个消息的主题。因此，服务高度依赖于要直播的主题。但是，在许多情况下，我得到了关于主题的leader not available、broker not available和leader= - 1。有人能对这些主题的可靠性和可靠性有所了解吗?如果我们能够解决上述问题，我们还能恢复吗？
浏览 0提问于2018-01-24得票数 5
回答已采纳
1回答
在文档中，我仍然不明白如何在broker.xml中配置STOMP协议的消息过期时间。我们有一个聊天项目，其中“多播”地址被创建为每个聊天 topic/chat.1, topic/chat.2, ..., topic/chat.100000 在其他方面，我们使用"Durable Subscriptions因此，即使是在我们不受欢迎的项目中，每月的</em
浏览 26提问于2020-09-03得票数 0
回答已采纳
，blocked属性将被设置为True，并在发布消息后等待while循环。在收到消息返回时，我会将blocked属性设置为False，循环将结束，并将消息返回给后端视图，返回错误或正确的消息。 问题是，从后端发布消息永远不会到达仿真器MQTT客户端。我不知道为什么，但在我的循环中是超时(10s)，在这个时间之后，我应该抛出模块没有响应的错误。我非常小心地调试了整个过程，当后端抛出错
浏览 66提问于2020-04-13得票数 0
我已经使用WSO2消息代理创建了一个示例发布和订阅模型项目。("publish data") publish_1(client,topic)
topic_pub= on_message
thread1=threading.Thread(target=publish_
浏览 4提问于2017-03-29得票数 0
2回答
问题是:在这个设计中是否需要一个恢复计划，卡夫卡会不会因为崩溃、磁盘故障而丢失数据？或者卡夫卡有自己的恢复机制，所以用户不需要自己的恢复计划？
浏览 0提问于2018-04-06得票数 3
回答已采纳
2回答
我有一个运行DCOS的小集群。我能够按照指南成功安装kafka。/kafka-console-producer.sh --broker-list 10.131.17.126:9475, 10.131.24.6:9655, 10.131.14.192:9181 --topic(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
我怀疑这可能与私有ip地址与公共ip地址以及server.properties中</
浏览 8提问于2016-09-19得票数 0
1回答
转到脚本：问题是在主题中找到当前消息的实际数量。我可以用以下方式显示消息总数： --broker-list $KAFKA_BOOTSTRAP--topic $TOPIC --time -1
但是，这包括已过期的消息
浏览 1提问于2018-05-10得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
腾讯会议活动推荐
腾讯云开发者
