当我使用kafka-clients 2.2.0制作和消费kafka消息时，我的producer运行良好，但消费者无法获得任何消息。但是当我使用shell时，这个命令不能获取消息： /opt/kafka_2.12-2.2.0/bin/kafka-console-consumer.sh --bootstrap-server 10.2.0.242:9092 --topic mytesttopic --from-beginning 此命令可以</e
浏览 48提问于2019-05-30得票数 0
来生成和消费消息，并且成功了。我用producer API编写了一段java代码来生成消息，这是可以的。通过kafka-console-consumer.sh验证。，但消费者代码不能工作。示例消费消息成功，消息由producer制作为topic test：为什么</e
浏览 9提问于2017-07-09得票数 4
我尝试使用以下命令在kafka consumer中读取消息：这里我们可以读取大约4天的旧消息，因为我们在kafka服务器配置文件中设置了保留时间为7天。但是，当我们尝试使用kaka-python客户端库的KafkaConsumer读取消息时，如下所示：
cons = KafkaCo
浏览 2提问于2015-09-01得票数 0
我可以使用Kafka CLI工具确认它确实是登陆数据。我每秒都会收到多条消息。kafka-console-consumer.sh --zookeeper svr:2181 --topic test(Pdb) consumer.poll(
浏览 18提问于2017-03-15得票数 2
1回答
但是，该主题是正确的，我可以看到使用控制台使用者将事件发送到主题：当我使用调试器并调用kafkaConsumer.listTopics()时，我还在下面的Scala代码示例中看到了这个主题而且，这是从单个单元测试中调用的，因此我只创建这个特性和使用者
浏览 0提问于2018-12-20得票数 2
回答已采纳
1回答
一旦它加入，它将使用来自主题的关于它需要执行什么计算的消息，然后开始计算。现在，由于这个使用者正在执行一个计算任务，我想让它退出使用者组，因为它没有执行新计算的任何进一步能力。在卡夫卡能做到这一点吗？
浏览 0提问于2019-09-05得票数 0
回答已采纳
然后我运行这个命令我可以使用下面的命令在消费者中获取消息bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic replica
但是我不能用pythonreplica', value=dat
浏览 0提问于2019-05-20得票数 0
我有一个docker容器运行在与--network=host相同的本地机器上，我希望将消息从docker容器内部发送到我的本地kafka代理。在docker内部，我可以通过运行以下命令连接到zookeeper并查找现有主题：./bin/kafka-topics.sh --zookeeper localhost:2181 --list
但是我既不能用kafka-console-consumer.sh也不能用kafka-console-producer.sh连接kafka
浏览 10提问于2019-10-18得票数 2
我启动了两个消费者来使用来自Kafka的消息，有六个分区，所以每个消费者被分配了三个分区。我发现消费者在消费一段时间后无法轮询任何消息，我可以看到滞后的存在，这意味着仍然有未轮询的消息。当我终止程序并重新启动它时，消费者可以使用剩余的消息，但过了一段时间，它再次停止，尽管还有剩余的消息。没有任何错误信息，请将我的部分代码如下：
KafkaConsumer<String, byte[]> consum =
浏览 3提问于2018-02-22得票数 0
我正在尝试探索KafkaConsumer是否可以在消息到达某个主题时接收消息，而不是在特定的时间间隔轮询。我知道Kafka是消费者端的一种拉取机制，由于我不确定消息何时到达，也不确定轮询时间间隔的最佳值，是否可以达到上述要求？
浏览 0提问于2018-08-07得票数 0
dcos kafka topic create topic1 --partitions 3 --replication 3Producer正在工
浏览 1提问于2017-02-28得票数 8
回答已采纳
我有一个Akka Actor，我想向它发送“控制”消息。这个Actor的核心任务是监听Kafka队列，这是一个循环中的轮询过程。我发现以下代码只是简单地锁定了Actor，它不会收到“停止”(或任何其他)消息： private var done = falsedef receive = { done = true &#x
浏览 3提问于2016-07-21得票数 2
您好，我一直在努力学习卡夫卡，并与我的远程轮询者/消费者有问题。bin/kafka-console-producer.sh --broker-list localhost:9092 --topictest
bin/kafka-console-consumer.sh --bootstrap-ser
浏览 3提问于2017-06-17得票数 0
我们尝试在指定的窗口时间从Kafka读取数据(所以我们有Kafka消费者)，这意味着避免在其他时间读取数据。然而，我们不确定在时间段到期后如何关闭消费者。我想知道有没有这样做的例子？非常感谢您的帮助。
浏览 34提问于2019-12-10得票数 1
回答已采纳
我可以像文档中描述的那样从kafka控制台消费消息。但是，当我在本地使用Python创建kafka消费者时，我无法消费消息。应该注意的是，kafka控制台工作得很好！我尝试调查此问题，但无法获得类似的环境/情况from kafka import KafkaConsumer
consumer = KafkaConsumer('dbserver1我确信这些消息是存在的，因为当我打开一个控制台使用者时，我可以</e
浏览 1提问于2019-04-29得票数 0
1回答
我正在寻找一个支持延迟消息传递的分布式排队系统，也就是说，从生产者发送消息时起，用户只能在指定的延迟后才接收消息。是我最熟悉的系统(特别是特性)，但我所处的环境是禁止使用AWS的。有可能使用作业调度程序而不是消息队列，但对于我的情况来说，这似乎是一个非常糟糕的阻抗错配。
我可以采用以数据库为队列的方法，但我不想这样做有各种各样的原因。
浏览 2提问于2018-11-06得票数 1
1回答
我试图以以下方式手动偏移提交：val sender: KafkaSender
.flatMap { Flux.fromIterable() }
sender.send(flux) // sending Flux t
浏览 11提问于2022-10-26得票数 0
我得到了两个不同的结果：
如果我先进行分区，然后生成消息并调用方法。KafkaConsumer可以成功地轮询未提交的消息。但是，如果我首先生成消息，而没有提交它的偏移量，那么在几秒钟后暂停分区，以调用恢复方法。KafkaConsumer将不会接收未提交的消息。我需要一些建议，或者有人可以告诉我卡夫卡最初的机制。
浏览 0提问于2018-11-12得票数 0
回答已采纳
云服务器
ICP备案
云直播
对象存储
实时音视频
腾讯云开发者
