我有一个接收二进制数据(原始数据包捕获数据)的Kafka主题。我可以使用Kafka CLI工具确认它确实是登陆数据。我每秒都会收到多条消息。kafka-console-consumer.sh --zookeeper svr:2181 --topic test
但是当我使用kafka-python时，我无法检索任何消息。consumer = kafka.KafkaConsumer("test", bootstrap_servers=["svr
浏览 18提问于2017-03-15得票数 2
1回答
我已经使用kafka-python库编写了一个python脚本，它将消息写入和读取到kafka中。我编写消息没有任何问题；我可以使用kafka控制台工具检索它们。但是我不能使用我的python脚本来读取它们。我在我的消费者上有一个for，它在迭代的第一行冻结，永远不会返回。下面是我的代码：
consumer = KafkaConsum
浏览 12提问于2020-03-02得票数 1
回答已采纳
正如标题中所说，我想在我的主题中获得一些记录，但我找不到使用kafka-python库的解决方案。有谁知道吗？
浏览 88提问于2020-10-07得票数 3
1回答
我正在通过docker使用kafka环境。它正确地上升了！我正在尝试阅读在流媒体上收到的所有消息！
有什么改正的建议吗？
浏览 100提问于2021-05-18得票数 0
回答已采纳
样本代码： bootstrap_servers=config["kafka"]["brokers"].split(','), ena
浏览 3提问于2021-04-14得票数 0
我想使用KafkaConsumer从kafka-python到使用者主题中的第一个N条消息：import json
bootstrap_servers在本主题中有32个分区(从0索引到31)。从一开始，什么才是正确的语法？
浏览 1提问于2021-10-25得票数 0
回答已采纳
使用Apache的最新版本(0.8.1.1.)使用kafka-python 0.9.2从pypi安装(最近一次上传是在2014-08-27)，这与github上当前的主分支不同。当使用"SimpleConsumer“=>进行测试时，脚本崩溃和重新启动将消耗来自上一个已知偏移量的消息。当使用"MultiProcessConsumer“=>进行测试时，=>崩溃和重新启动时，脚本将从偏
浏览 5提问于2015-01-05得票数 0
4回答
场景： 我正在将数据JSON对象数据写入kafka主题，而读取时我只想读取基于消息中存在的值的特定消息集。我使用的是kafka-python库。示例消息： {flow_status: "completed", value: 1, active: yes}
{flow_status:"failure",value 2, active:yes} 在这里，我只想读取flow_Status为已完成的<e
浏览 78提问于2019-02-18得票数 2
回答已采纳
我正在寻找一种方法来显示我的API (本地主机)到我的docker使用kafka。producer.send('test', json.dumps(res.text).encode('utf-8'))#producer.flush()然而，我的消费者不能工作这是我到目前为止所尝试过的。当前在for循环
浏览 8提问于2021-04-03得票数 0
我正在使用kafka-python-2.0.1来消费来自kafka代理的数据。到目前为止，我只运行了一个消费者实例。我们每5分钟就会收到2M条记录。我注意到kafka-python不能以更快的速度读取数据以及时消费所有消息。我是kafka-python的新手，不确定如何实现才能更快地读取数据。我应该运行多个使用者吗？
浏览 31提问于2020-11-06得票数 0
1回答
概述:我正在开发一个聊天机器人，我希望存储关于每个用户帖子的数据，并能够访问它，以帮助制定对下一个帖子的响应。特定问题：是否有一种在C#中创建存储类的方法，以便在执行异步任务期间执行以下操作，该任务一直在等待用户输入：
接收新<e
浏览 2提问于2016-12-01得票数 0
回答已采纳
我有一个运行并经过测试的Kafka集群，并且正在尝试使用Python脚本向代理发送消息。当我使用Python3 shell并调用producer方法时，这是可行的，但是当我将这些相同的命令放入Python3并执行它时--脚本似乎挂起了。我正在为消费者和生产者使用kafka-python库。当我使用Python3 shell时，我可以看到使用Kafka工具2.0.4出现在主题中的<e
浏览 0提问于2019-04-05得票数 1
回答已采纳
1回答
从消息中心的主题中检索消息
、、、、
我正试图使用凝聚式Kafka Python从bluemix上的消息中心上获取来自主题的消息。我的代码在下面找到，但有些东西不起作用。主题和消息中心已经启动并运行，所以代码中可能有一些东西。所以我想这要么是连接失败，要么是无法检索消息。证书本身是正确的。
浏览 5提问于2018-03-16得票数 2
回答已采纳
我将在我的应用程序中使用Kafka作为消息代理。此应用程序完全使用Python编写。对于这个应用程序的一部分(登录和身份验证)，我需要实现一个请求回复消息系统。换句话说，生产者需要同步地从使用者那里获得产生的消息的响应。使用Kafka及其Python库(kafka-python，.)是否可行？
浏览 0提问于2020-11-09得票数 4
2回答
这是我的代码，它使用kafka-python。sasl_plain_password=SASL_PASSWORD)
print(msg)
我希望在循环中从now和month_ago之间的主题中获得结果谢谢你的帮助！
浏览 9提问于2022-03-09得票数 1
回答已采纳
有没有办法在不安装主包的情况下安装setup.py的install_requires中列出的所有包？该列表如下所示： 'cassandra-driver==3.23.0', 'toolz==0.9.0'
浏览 2提问于2021-03-03得票数 0
我计划跳过话题的开头，只阅读某个时间戳到结尾的消息。对于如何做到这一点，有什么建议吗？
浏览 3提问于2017-09-25得票数 2
回答已采纳
2回答
我是Python的新手，从Kafka开始。我有一个需要发送和使用json消息的需求。为此，我使用kafka-python与卡夫卡进行交流。)))for message in consumer :
print(message) 然而，我得到了以下关于消费者的异常
浏览 30提问于2019-04-05得票数 3
回答已采纳
使用kafka-python消费kafka，seek方法重置的本地偏移量会提交给kafka吗？我正在研究获取双中心机房Kafka集群的rpo索引的解决方案。使用kafka-python获取Kafka集群的最大时间戳，取两个机房的Kafka集群的最大时间戳之差。使用seek()将偏移量重置为分区的最大偏移量-1，然后使用poll()获取最新的<em
浏览 8提问于2021-05-20得票数 0
1回答
我试图用kafka-python构建一个应用程序，在这个应用程序中，使用者从一系列主题中读取数据。非常重要的是，消费者不会两次阅读同一条消息，但也不会错过一条消息。我只能读取主题中的所有消息(这会创建双读)，或者只侦听新消息，只监听(并且错过了在故障期间发出的消息)。当暂停消费者时，我不会遇到这个问题。
为了解决这个问题，我创建了一个孤立的模拟。如果auto_of
浏览 2提问于2018-08-11得票数 9
回答已采纳
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
