我正在尝试使用NET CORE 2.1制作一个消费者Kafka，这个消费者应该读取一条消息来比较时间戳和提交与否，所以这个消费者可以停留在相同的消息上，直到这个验证为真。Console.WriteLine($"Error occured: {e.Error.Reason}"); }
因此，我的主题中有10条消息我希望只有当我提交()前一条<e
浏览 10提问于2019-07-23得票数 0
1回答
我用卡夫卡和弹簧启动和弹簧卡夫卡。在异常的应用程序终止并重新启动之后，我的应用程序开始接收来自Kafka队列的旧的、已经处理过的消息。spring.kafka.bootstrap-servers=${kafka.host}:${kafka.port}spring.k
浏览 0提问于2018-07-08得票数 4
回答已采纳
1回答
嗨，我试图建立一个分析引擎，以确定客户端正在使用的urls\事件的实时分析，以及记录api的性能。以下是我计划实现的逻辑：
kafka代码过滤器创建一个过滤器来拦截url 代码过滤器作为一个可重用的jar，它具有使用kafka拦截器拦截它们的逻辑，如果url模式匹配，它将生成事件并将其发布到kafkastreams中。
浏览 0提问于2017-05-30得票数 0
我们有Kafka设置，以便能够通过多个服务器并行处理消息。但每条消息必须只处理一次(并且只由一台服务器处理)。我们已经启动并运行了它，它工作得很好。现在，我们面临的问题是，Kafka的消费者为了获得最大的效率而批量阅读消息。这会导致一个问题，如果/当处理失败时，服务器关闭或其他什么，因为我们丢失了即将处理的数据。有没有办法让消费者一次只读消息，让Kafka保留未处理的消息？例如；Consumer拉出一条消息
浏览 0提问于2015-08-17得票数 13
主题设置(我不太清楚每个选项的应用位置和前缀，因此可能有几个未使用和无关的选项以及夸大的数字-欢迎更正)： .via(producerFlow) .runWith(Sink.ignore)在等待了几秒钟之后ids.length消息，但它总是在第一
浏览 2提问于2017-06-22得票数 1
1回答
嗯，我在Spring和Confluent.Kafka 1.1.0 (.NET核心2.1)中做了一个性能测试。我注意到Spring在11秒内写了100万条消息，NETCORE在1.58分钟内写了100万条消息。两者之间的差别比预期的要大得多，在这两种生产者配置中，我没有改变任何东西，只使用默认的驱动程序。, "localhost:9092"}
浏览 0提问于2019-07-19得票数 1
回答已采纳
我正在使用Kafka2.1.0，希望在订阅完成后发布一条消息。是否有方法让生产者知道订阅是否已经发生，然后发布消息？否则我每次都会失去第一条信息。
浏览 0提问于2019-02-19得票数 0
回答已采纳
我试图使用Kafka在MongoDB和Elasticsearch之间建立一条简单的管道。插入的数据被成功地存储在Elasticsearch中，但是当我编辑或删除文档时，我只会得到另一个存储在Elasticsearch中的文档。fullfillment.elasticsearchApp.subscriber","schema.ignore": "true","behavior.
浏览 4提问于2020-04-28得票数 0
回答已采纳
我无法让下面的代码工作：var conf = require('./providers/Config');
let consumer = new kaf
浏览 1提问于2016-03-08得票数 3
回答已采纳
1回答
我正在学习kafka，我想把我的应用程序分成两个微服务。首先，将所有传入的消息从KafkaConsumer保存到数据库，然后根据给定的id选择entity。第二，提供REST来保存和获取实体。如何使用kafka从REST中的db接收存储的ID？这是在POST请求时调用生产者的示例代码。String, byte[]> result = resultFuture.get();}
以及将myObj
浏览 2提问于2017-11-28得票数 3
回答已采纳
我正在使用segmentio/kafka-go客户端来读取主题中的消息。我找不到..如何从上一条/新消息开始阅读。
每次启动代码时，它都会从该分区中的起始偏移量开始读取。
浏览 8提问于2019-11-24得票数 2
5回答
如果我在我的storm拓扑中增加了Kafka spout的并行度，我如何才能阻止它多次阅读同一主题中的同一消息？
浏览 2提问于2015-02-19得票数 1
1回答
我认识到，在Kafka中，每个分区的顺序是可以保证的。但是，如果有多个分区，而生产者没有指定键，而只有一个使用者(为什么有一个使用者)，那么分区会受到怎样的影响？对于当前的数据负载1是很好的，有多个分区供将来使用)1 consumer3)即使我们指定了分区密钥，我们是否可以保证我们的顺序已经到
浏览 0提问于2018-11-26得票数 1
1回答
卡夫卡有重复信息
、、
在生成或使用数据时，我没有看到任何失败，但是在生产过程中有很多重复的消息。对于一个获得大约100 k消息的小主题，有~4k重复，但正如我所说的，没有失败，而且没有实现重试逻辑，也没有设置配置值。我还检查那些重复消息的偏移值，每个值都有不同的值，这告诉我问题在生产者中。
任何帮助都将不胜感激。
浏览 1提问于2015-12-02得票数 5
我们使用Kafka来存储由集群中的节点生成的消息，并分发到集群中的所有节点，我让它主要使用akka-streams，但有几个问题我必须解决。这是有一些限制的。首先，消息必须由集群中的每个节点使用，但只由一个节点生成。我知道我可以为每个节点分配一个组id，这可能是它的节点ID，这意味着每个节点都会收到消息。已经排序好了。但这里有一些问题。数据是非常短暂和相当大的(仅低于一兆字节)，不能进一步压缩或分解
浏览 0提问于2017-12-06得票数 0
lines = Encoding.UTF8.GetString(message.Value); }
}但是，如果您看到文本文件，
浏览 0提问于2018-04-27得票数 3
回答已采纳
1回答
可以配置kafka消费者，并在调试时确认其在每次轮询kafka主题时消耗数千条消息。 @Override
public void c
浏览 11提问于2022-06-26得票数 0
1回答
我使用的是合流3.3.0。我使用jdbc-source-connector将消息插入到我的Oracle表中的Kafka中。这个很好用。
我想检查一下是否可以“重新插入”。我的意思是，如果我有一个学生表，有3列id(数字)、name(varchar2)和last_modified(时间戳)。每当我插入新行时，它都会被推送到Kafka (使用timestamp+auto增量字段)。但是当我更新行时，应该更新Kafka<e
浏览 1提问于2018-08-01得票数 0
我想在我的项目中为实时消息处理实现Storm。我观察到很多人使用“Apache”和“Storm”。在我的项目中，客户端应用程序将向服务器端发送消息，服务器端应该对消息进行身份验证、处理并存储到HBase中。唯一的限制是消息不应该被丢弃，每条消息都必须持久化到HBase中，如果处理该消息只需几分钟就可以了。
在Sto
浏览 0提问于2015-07-01得票数 0
回答已采纳
1回答
我有一个kafka用户，它使用来自某个主题的消息并对其进行处理。我已经应用了5的重试尝试来处理该消息并将该消息写入另一个主题以供将来参考，以防在5次尝试之后处理失败。}")
@KafkaListener(topics = "${kafka.consumer.topic}", groupIdkafkaTemplate.send(failedTopic, kafk
浏览 11提问于2020-07-13得票数 0
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
