RabbitMQ是否支持生产者端批量发送消息，消费者端批量消费消息？就像Kafka一样，它存储批量产生的消息，直到达到linger.limit，或者直到批量满！RabbitMQ是否也支持生产者端的批处理？
浏览 7提问于2017-06-15得票数 2
回答已采纳
2回答
寻找关于我想要对主题中的消息进行处理的想法。我希望能够处理消息，在我的情况下，以10,000个为一批的事件。这是因为我在转换后将消息插入到我们的雪花仓库中。雪花加载在批处理加载时执行得更好。在主题中有10,000条消息之后才从主题中拉出消息的消费者有什么想法吗？IOW，一旦延迟达到10000，就从主题中拉出消息。有没有关于如何建立这样一个消费者的想法？
浏览 23提问于2019-08-08得票数 0
高级消费者API似乎是一次读取一条消息。如果消费者想要处理这些消息并将其提交给其他下游消费者，比如Solr或Elastic-Search，这对他们来说可能是一个很大的问题，因为他们更喜欢批量发送消息，而不是一次只有一个。在内存中对这些消息进行批处理也不是一件容易的事，因为Kafka中的偏移量只有在批处理已经提交时才需要同步，否则具有未提交下游消息的崩溃的kafka消费者(如Solr或ES)的偏移量将
浏览 2提问于2016-03-04得票数 9
我在我的笔记本电脑上设置了一个小测试，它执行以下操作：然后，我将这些列应用于只包含一条消息的迷你批处理。writeStream
.format("console"
浏览 0提问于2019-10-31得票数 3
我们有Kafka设置，以便能够通过多个服务器并行处理消息。但每条消息必须只处理一次(并且只由一台服务器处理)。我们已经启动并运行了它，它工作得很好。现在，我们面临的问题是，Kafka的消费者为了获得最大的效率而批量阅读消息。这会导致一个问题，如果/当处理失败时，服务器关闭或其他什么，因为我们丢失了即将处理的数据。有没有办法让消费者一次只读消息，让Kafka</em
浏览 0提问于2015-08-17得票数 13
2回答
我们需要对elasticsearch执行批量写入。我们想知道是否有更好的方法对数据进行批处理，并在进行批处理时避免数据丢失。initialize(topic, consGroup, STREAMSERDE.STRINGDESER, STREAMSERDE.JSONDESER, props);
logger.debug("Kafka
浏览 1提问于2019-01-22得票数 0
1回答
listenSingle(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) { messages.forEach({ });根据文档，它似乎对卡夫卡消费者没有任何影响(无论如何，卡夫
浏览 2提问于2019-12-05得票数 6
我有一个可以批量消费消息的Kafka监听器。我也在使用手动提交。如果其中任何一条消息出现问题，我将使用返回语句，而不是提交偏移量(我希望再次处理所有这些消息)。我的问题是- 在Kafka Listener中使用return语句有副作用吗？再次消费相同的消息需要多长时间？ 我正在使用Spring Kafka和SpringBoot。
浏览 34提问于2021-04-26得票数 0
回答已采纳
我将数据从数据库导出到Kafka中，方法是以固定大小的批量发送数据。如果有一个批次失败，我只会重新发送整个批次。我的想法是，而不是找出哪些消息被正确导入，以处理某些消息被导入两次的事实。Kafka实现了这个压缩步骤，在所有具有相同消息键的消息中，只保留最新的消息。压缩将删除所有重复的消息，因为所有消息都具有唯一的关键字。
有没有办法只读Kafka主题，直到压缩完成？如果我不能一直读
浏览 4提问于2017-12-06得票数 0
我正在设计一个apache storm拓扑(使用streamparse)，它使用一个spout (apache kafka spout)和一个并行大于1的螺栓构建，可以批量读取kafka spout中的消息并将消息持久化到mysql表中当螺栓在mysql上插入失败时，我不会在kafka中提交偏移量，但一些消息已经在spout发送到螺
浏览 2提问于2019-01-31得票数 0
1回答
我想批量消费一个Kafka主题，我想每小时阅读一次Kafka主题，并阅读最新的每小时数据。val readStream = existingSparkSession .format("kafka") .option("subscribe", "kafka.raw")但
浏览 4提问于2021-03-10得票数 0
还请注意，可靠性是我的第一要务，因为数据不能丢失，而且应该有一种机制在代码中返回响应，以便处理成功或失败。使用http似乎是健壮的，并且有基于的确认机制，但是如果采用这种方法，在scala中的http客户机库更合适，因为我需要以键值格式的顺序发送大量数据，并以无阻塞的方式处理响应？
浏览 2提问于2019-05-08得票数 1
回答已采纳
我已经在生产者端启用了快速压缩，批处理大小为64kb，并处理每个1KB的消息，并将逗留时间设置为inf，这是否意味着直到我处理64条消息，生产者才会将消息发送到kafka out topic...换句话说，producer会将每条消息发送给kafka，还是等待64条消息并分批发送……
因为偏移量是一个接一个地增加，而不是64的倍数。编辑-使用flink-kafka连接器
浏览 0提问于2018-08-10得票数 1
kafka是否支持(或计划)在批处理中使用配置了死信队列的粒度。说批次大小为100 消费者在一个中成功地读取了100条日志记录，除了批量订单20、51、99的记录外
kafka是否允许发送表单的压缩(19，-1,30，-1,48，-1,1)，从而将具有批处理索引20、51和99的消息发布到死信队列中。
浏览 0提问于2021-03-26得票数 3
2回答
下面是我用来发送消息的python生成器代码。我首先启动了Kafka控制台使用者，然后运行python代码。from confluent_kafka import Producerfrom faker import Faker
producer.produce(topic="first_topic",value=product.serial()) 我面临的问题是，如果我在调用produce之后调用flush方法，消息</
浏览 37提问于2020-11-20得票数 0
我需要使用consume process-produce模式来处理Kafka消息，并配置了一个带有Kafka事务管理器的Spring Kafka侦听器容器，还设置了transaction-id-prefix我使用了批处理的ack-mode，并试图了解在这种事务模式下，何时会实际提交偏移量。文档似乎表明，一旦使用了轮询中的所有记录，ack-mode批量提交偏移量-在事务上下文中也是如此，即每个轮询1个事务？或者在使用Kafka事务时忽略了ack-mode
浏览 20提问于2020-03-12得票数 0
回答已采纳
1回答
批处理模式下的错误处理
、、、、
我正在使用和Kafka来批量使用来自Kafka主题的消息。我正在尝试实现一个错误处理机制。根据我的理解，我不能在批处理模式下使用Stream的enableDLQ属性。我发现RecoveringBatchErrorHandler和DeadLetterPublishingRecoverer可以重试并发送spring文档中的失败消息。
浏览 5提问于2022-07-06得票数 0
回答已采纳
("CAST(id AS STRING) AS key", "to_json(struct(*)) AS value") .option("topic", "testing")
.option("kafka.bootstrap.servers",class.foreach(Re
浏览 3提问于2018-09-12得票数 1
我希望以固定的时间间隔从Kafka主题获取一批消息(比如10000条消息)(用于一次性批量处理所有10k条记录，而不是逐条处理)。
浏览 14提问于2017-12-15得票数 1
回答已采纳
我需要通过调整批量大小来减少Kafka producer发送到Kafka集群的请求数量。我使用的是2分钟的批量处理预期结果：
我需要测量Kafka producer向Kafka集群发送记录的呼叫或通话次数。如果Kafka在集群上花费了更多的时间，那么我可以调优linger.ms和批处理大小
浏览 2提问于2019-08-28得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
