无法使用Kafka消费大消息(超过1.3MB)？
Kafka是一种分布式流处理平台，用于构建高性能、可扩展的实时数据流应用程序。它通过将数据分割成多个分区并在多个服务器上进行分布式存储和处理，实现了高吞吐量和低延迟的数据传输。
当无法使用Kafka消费大消息（超过1.3MB）时，可能是由于以下原因：
- 消息大小限制：Kafka默认情况下对消息大小有一定限制，超过该限制的消息将被拒绝。这个限制可以通过配置文件中的
message.max.bytes参数进行调整。如果消息超过了该限制，可以尝试增加该参数的值。 - 分区大小限制：Kafka将消息分割成多个分区进行存储和处理，每个分区都有一定的大小限制。如果消息超过了分区的大小限制，将无法成功写入或消费。可以通过调整分区的大小限制来解决该问题。
- 网络传输限制：Kafka在网络传输消息时也有一定的大小限制。如果消息超过了网络传输的限制，可能会导致消费失败。可以尝试增加网络传输的限制或者考虑使用其他方式进行大消息的传输。
针对以上问题，腾讯云提供了一系列解决方案和产品，例如：
- 腾讯云消息队列 CMQ：腾讯云消息队列 CMQ 是一种高可用、可伸缩、可靠的消息队列服务，支持大消息（最大2GB）的传输和消费。您可以使用 CMQ 来替代 Kafka，以满足大消息的需求。了解更多信息，请访问：腾讯云消息队列 CMQ
- 腾讯云云原生数据库 TDSQL-C：腾讯云云原生数据库 TDSQL-C 是一种高性能、高可用的云原生数据库，支持分布式事务和分布式消息队列，可以满足大消息的存储和消费需求。了解更多信息，请访问：腾讯云云原生数据库 TDSQL-C
请注意，以上产品仅作为示例，具体的解决方案和产品选择应根据实际需求和场景进行评估和选择。
相关·内容
我按照链接解决了kafka msg限制问题。但没那么走运
我试着增加
A.)在代理上：
message.max.bytes=15728640
replica.fetch.max.bytes=15728640
B.)关于消费者：fetch.message.max.bytes=15728640
仍然面临着同样的问题。特定主题不能消耗超过1.3MB的数据
在我的应用程序中，从python代码中发送一个主题的msg，并在节点服务器上使用它。
浏览 0提问于2018-08-18得票数 0
1回答
在分区0上测试一条Kafka消息，处理时间超过一个小时。我尝试过自动提交，暂停/恢复，增加最大轮询超时，但我无法停止消费者消息复制。这是使用python的融合式kafka。我看不出有什么方法可以实验异步/同步提交。
msg= self.consumer.poll(timeout=1.0)
if msg:
self.consumer.pause()
some task.. (takes ~60 minutes minutes)
self.consumer.res
浏览 3提问于2022-03-07得票数 0
我正在运行kafka_2.11-0.9.0.0和一个基于java的生产者/消费者。消息大小约为70KB时，一切工作正常。然而，在生产者将更大的70MB消息排入队列后，kafka似乎停止向消费者传递消息。即不仅不递送大消息，而且还递送随后的较小消息。我知道生产者成功了，因为我使用了kafka回调进行确认，并且我可以在kafka消息日志中看到消息。
kafka配置自定义更改：
message.max.bytes=200000000
replica.fetch.max.bytes=200000000
消费者配置：
props.put("fetch.message.max.bytes"
浏览 1提问于2016-02-03得票数 0
2回答
假设我有3个Kafka经纪人，一个动物园管理员，50个生产者，50个消费者和1个主题(testTopic1)。并且所有的消费者都订阅了testTopic1。现在，我将与50个生产者同时向同一主题(testTopic1)发送50条消息。现在我希望Kafka集群不会同时向消费者发送超过40条消息。剩下的10个将保留在队列中或放弃它。
也许这是Kafka中的负载均衡。我不知道我将如何做这项工作。我是新来卡夫卡的，请帮帮忙。
浏览 17提问于2019-07-14得票数 0
似乎我的卡夫卡节点消费者：
var kafka = require('kafka-node');
var consumer = new Consumer(client, [], {
...
});
在某些情况下我处理不了太多的消息。是否有办法限制它(例如，每秒接受不超过1000条消息，可能使用暂停api)？
我使用的是kafka节点，与Java版本相比，它似乎有一个有限的api。
浏览 14提问于2016-07-23得票数 6
3回答
我有一个非常非常大(消息计数)的Kafka主题，它可能每秒有超过20M条消息，但是，消息大小很小，它只是一些纯文本，每个不到1KB，我可以使用几个分区每个主题，也可以使用几个服务器在一个主题上工作，他们将消耗主题中的一个分区...如果我需要+100台服务器来处理一个庞大的主题，该怎么办？
在单个主题上创建+100个或更多分区是否合乎逻辑？
浏览 2提问于2019-06-11得票数 0
我是Apache Kafka的新手，我想知道apache Kafka中的消息可以有多大。如果消息的大小变得相当大，比如说几百MB，使用Apache Kafka是否有效？
我有一个场景，我想要将文件复制到HDFS以供Hadoop作业使用，这些文件也由其他进程使用。我在考虑首先将文件复制到Apache Kafka中，然后消费者可以将它们复制到HDFS，而其他消费者可以使用Kafka。这是不是最好的方法？
浏览 0提问于2015-04-29得票数 1
是否有删除一个主题中超过一个小时的kafka消息的选项。只需要在一个主题中最后一个小时的消息。Kafka producer会在主题的每一分钟都接受消息。我需要使用当前主题的最后一个小时的数据。1.是否可以选择删除主题中时间超过一小时的消息，需要在主题中保留少于一小时的数据2.是否可以从主题中读取最近一小时的数据，主题中包含完整的一天数据
浏览 1提问于2018-10-27得票数 0
来自我的kafka生产者的一些消息超过了默认的1MB，并且被代理拒绝。
我无法访问代理和消费者来进行更改。因此，我只需要在生产者级别进行更改。
我正在考虑在我的producer中使用batch.size到5MB的snappy压缩。请让我知道这是否是正确的选择或任何其他方式可用于我的用例。谢谢
浏览 0提问于2020-07-23得票数 0
卡夫卡中的demo.incoming话题堆积如山。
我正在使用。每秒大约有70个请求通过Scrapy Cluster REST api (Producer)提交给Kafka。蜘蛛可以非常快地完成爬行，因为redis中的队列保持在一个非常低的数字，大多数时间少于10个。但demo.incoming中的消息数量每秒都在堆积。这是我用来查看Kafka中demo.incoming主题消息数的命令
kafka-run-class.sh kafka.tools.GetOffsetShell \
--broker-list localhost:9092,kafka-statefulset-2:9092,ka
浏览 1提问于2019-07-07得票数 1
我想知道spring boot内部的spring kafka Producer是否保证交付。
有人知道如果某个随机监听器无法接收消息会发生什么情况吗？spring kafka是否会重试发送消息？
浏览 0提问于2019-05-29得票数 0
1回答
当我检查使用者延迟时，它会显示某个特定的consumer-id是从某个主题的特定主机消费中运行的。但是当我去那个主机时，没有这样的消费者在运行。
如何杀死此consumer-id，以便可以为其所属的组重置使用者偏移量。
Kafka server version: 0.11.0.1
Kafka client version(stuck): 0.10.0.2
这个consumer-id被卡在了第一位，因为它无法使用消息，因为有些消息在Kafka中有标题。
我已经尝试过以下几种方法：
从不同的主机和不同的卡夫卡版本消费，它消费的信息，但消费者id，主机不会改变。
重新启动kafka代理，它
浏览 0提问于2019-02-18得票数 3
我可以创建一个kafka消费者来阅读固定数量(比如100条)的Kafka消息吗？我的条件是无论如何都不能超过100，小于100就可以了。
浏览 1提问于2018-10-25得票数 2
我正在尝试用kafka实现一个多消费者-单生产者的设置。我创建了一个包含3个分区的主题，如下所示：
./kafka-create-topic.sh --topic stream.main.out --zookeeper
localhost:2181 --partition 3
producer的设置如下：
props = new Properties();
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("broker.l
浏览 0提问于2018-03-23得票数 0
我有两台机器localhost和192.168.1.110来运行两台独立的单机kafka。
(1)在localhost上，我运行：
主控台kafka_2.11-0.10.0.0bin/kafka-➜-Producer.sh --broker-list 192.168.1.110:9092 --topic test这是一条消息2016-08-24 18:15:27,441向topic test发送消息时出错: null，value: 2字节错误:包含1条记录的(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) or
浏览 13提问于2016-08-24得票数 0
1回答
我有一个应用程序，它为REST请求提供服务器，并且正在侦听一个Kafka主题。我将应用程序部署到Kubernetes，并像这样配置就绪探测器
readinessProbe:
exec:
command:
- cat
- /tmp/healthy
initialDelaySeconds: 5
periodSeconds: 5
基本上按照的指示
部署完成后，我可以看到吊舱准备探测器失败了。
Readiness probe failed: cat: can't open '/tmp/healthy': No such file or dir
浏览 3提问于2019-11-21得票数 4
回答已采纳
我想确保我在Kafka中不会丢失任何消息，即使我的消费者宕机了。
如果我有一个小时的日志记录，如果我的消费者停机超过一个小时，我将丢失未使用的消息。
在Kafka 0.10.2.0中，有没有办法轻松地检查所有分区的所有滞后？
我发现很多方法都不能很好的工作。
浏览 14提问于2017-03-16得票数 7
我有一个队列(在本例中是Amazon SQS )，并且有N个相同服务的节点正在运行，这些节点正在使用来自SQS的消息。
如何确保在任意时间点内，不超过一个节点从队列中读取相同的消息。
在Kafka的情况下，我们知道，同一消费者组中不能有超过一个消费者被分配到单个主题分区。我们如何确保在Amazon SQS中处理相同的事情？
浏览 0提问于2019-04-01得票数 1
1回答
如果http请求超时，我正在尝试中断当前线程。我已经将PlatformTransactionManager设置为Kafka事务作为一个bean。我在方法级别使用了@Transactional注解。我们将在3个主题中发布消息。在第一个主题中发布消息后，我放入Thread.sleep(5000)，如果执行时间超过6秒，当前线程将从过滤器中断。因此，这里的通话被打断了，但消息正在发布到Kafka。我们只是在产生信息。我们没有消费任何消息，但能够在我们内部的Kafka检查工具中看到消息。我们使用KafkaTemplate.send()来发送消息。
浏览 49提问于2021-09-05得票数 0
我在一个kafka集群中创建了两个kafka brokers。当一个经纪人宕机时，我无法将任何数据提供给kafka消费者。我使用这个命令来读取来自消费者的消息：
bin/kafka-console-consumer.sh --topic test_kafka_cluster \
--bootstrap-server 127.0.0.1:9092,127.0.0.2:9092 --from-beginning
浏览 10提问于2021-01-22得票数 0
我有以下带有kafka输入的logstash配置
input {
kafka {
bootstrap_servers => "localhost:9092"
topics => ["mytopic"]
}
}
filter {
json {
source => "message"
}
}
output {
stdout {
codec => rubydebug
}
elasticsearch {
hosts => ["localhost:92
浏览 5提问于2017-02-04得票数 2
我使用Kafka Rest代理来生成和使用Azure APIM从Kafka主题消费，因此使用Rest代理的原因。我可以使用auto.offset.reset =最早的线性方式使用消息，所以每次调用Kafka Rest都会给我最新的消息，这是很棒的，
在以下情况下：
A1 is data requestor
B2 is API that fulfills A1's request using REST proxy for Kafka
C3 is Kafka Rest proxy server
如果A1向B2请求数据，而B2使用卡夫卡主题C3提供的消息给A1，但由于某些互联网问题，该消息
浏览 2提问于2022-07-21得票数 0
1回答
我有以下场景。 我们在K8s的云中对我们的应用程序进行了负载测试。我们的入站消息是通过Kafka发送的，我们正在给Kafka回信。我们的架构是这样的: Kafka线程将消息推送到disruptor (阻塞等待策略，512大小)，业务线程将消息从disruptor中取出进行处理。为了模拟负载，我们在Kafka主题(4个分区)中设置了应用程序未运行时的关闭500K消息。然后，我们启动应用程序来测量负载。 我们看到的是，disruptor完全填满了剩余的0容量，然后开始耗尽，这种情况一次又一次地继续下去。 这是正确的行为，还是我们以错误的方式使用颠覆者？请分享你的想法？
浏览 35提问于2020-07-21得票数 0
据我所知，kafka消费者只能使用API轮询()从代理中提取消息，但是我们知道我们也可以使用kafka高/low级别的使用者API从代理获取消息，那么kafka高/low级消费者API和API轮询()之间的关系是什么？高/low级别的使用者API调用轮询()内部？
浏览 1提问于2017-05-15得票数 0
回答已采纳
我在本地服务器上安装了kafka，通过在服务器上运行的一些其他应用程序，生产人员正在将消息发布到我的kafka服务器内的代理，通过zookeeper，我可以很容易地看到我的kafka的健康状况，它显示了在我的kafka服务器中创建的所有主题，主题中的偏移量等等，所以zookeeper唯一不能显示的是单个主题中的消息，所以有人推荐kafka-manager工具，我安装并运行它，它工作得很好，它显示了来自我的kafka服务器的许多信息。但它仍然不能显示由我的kafka服务器中的各个消费者发布或消费的真实消息，所以我的问题是，是否有方法/工具/代码来找出发布或消费的消息，我的意思是除了这个kafka
浏览 7提问于2016-08-19得票数 0
1回答
我可以针对Kafka location安装在命令行上发送和接收消息。我还可以通过Java代码发送消息。这些消息显示在Kafka命令提示符中。我还有一个用于Kafka消费者的Java代码。代码昨天收到消息。然而，它今天早上没有收到任何消息。代码未更改。我想知道属性配置是不是不太正确。下面是我的配置：
制片人：
bootstrap.servers - localhost:9092
group.id - test
key.serializer - StringSerializer.class.getName()
value.serializer - StringSerializer.class.ge
浏览 4提问于2016-06-24得票数 0
我的生产者从一条输入消息生成n任务，并在topic上发布这些任务。
要求是，在topic的消费者组中的所有单个消费者中，没有人在1小时内处理这些n任务中的任何一个任务超过3个。
这意味着如果我想立即处理所有这些消息，我至少需要ceil(n/3)使用者。如果少于ceil(n/3)的消费者，那么我需要一些方法来延迟消息，直到在最后一个小时内有num_processed < 3。
在实现这个解决方案的实用性方面，我希望将Kafka与Faust 1一起使用，但如果需要，我也可以访问Redis。
到目前为止，我的想法是确保在生产时至少有ceil(n/3)消费者，然后由生产者将任务循环分配给topi
浏览 4提问于2020-01-24得票数 0
librdkafka包含函数rd_kafka_position，它获取给定主题-分区的当前偏移量。但评论说：
每个请求分区的\p偏移字段将被设置为上次使用的消息+ 1的偏移量，如果没有先前的消息，则设置为RD_KAFKA_OFFSET_INVALID。
换句话说，如果还没有使用任何消息，它就不会给您提供任何有用的信息。
我对我刚刚订阅了一个主题的情况很感兴趣，并且我已经打电话给rd_kafka_seek了：
寻求一个已知的位置(在错误恢复的情况下)，或者寻求分区的最末端。
在这种情况下，我想知道的是，如果一个消息被消耗，那么下一个消息的偏移量是多少。换句话说，在第一种情况下，它应该是传递给r
浏览 6提问于2020-07-24得票数 2
1回答
我对Kafka非常陌生，我们使用的是Kafka 0.8.1。
我需要做的是从topic消费一条消息。为此，我必须用Java编写一个使用者，它将使用topic中的消息，然后将该消息保存到数据库中。保存消息后，将向Java使用者发送一些确认消息。如果确认为true，则应从主题中使用下一条消息。如果acknowldgement为false(这意味着由于从主题中读取的某些错误消息，无法保存到数据库中)，则应再次读取该消息。
我认为我需要使用Simple Consumer来控制消息偏移量，并且已经完成了这个链接中给出的简单的消费者示例。
在这个例子中，偏移量在run方法中被计算为'readOff
浏览 2提问于2014-07-31得票数 3
回答已采纳
我有一个kafka消息队列，我无法访问kafka安装的bin文件夹。
在这种情况下，是否可以删除队列中的所有消息？创建一个客户端并仅使用所有消息是否足够？
浏览 0提问于2020-11-13得票数 0
我有一个独立的Kafka设置与单盘。计划流式传输超过百万条记录。如何为我的主题确定分区以获得更好的吞吐量？必须是1个分区？
在独立Kafka服务器上，是否建议一个主题有多个分区？
浏览 2提问于2021-03-30得票数 0
如果我运行的Kafka集群的分区数超过了我唯一的消费者组的消费者数量。对于消息的排序，或者跨分区的消息的及时传递，有什么保证吗？
简单的例子：
2个分区，1个消费者
生产者通过一个键控制分区分配。
消息1进入并去往分区A
消息2进入并去往分区B
消息3进入并去往分区A
我知道消息1将在消息3之前被消费，因为它们在同一个分区中。但是消息2呢？是在Message 3之前还是之后消费？或者可能会有所不同？它可能在消息1之前被消费吗？
此外，如果分区A的新消息不断传入，并且生产速度快于消耗速度，该怎么办？消息2是否会无限期地驻留在分区B中？什么时候会被消费？有没有保证消息不会永远留在那里？
更一般的情
浏览 0提问于2014-01-23得票数 17
回答已采纳
在上查看此文档
Zookeper在端口2181 Broker/Kafka启动，时间为9092。
启动生成器发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka // producer is producing directly to broker not zookeeper which is correct
启动使用者接收消息
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-
浏览 15提问于2017-07-11得票数 3
回答已采纳
所有的人，请原谅我，我是一个新手刚刚开始的Kafka。目前，我正在阅读Kafka关于的文档。
如文件所述。
对于传统的信息系统。它们不能缩放消息处理。因为
发布-订阅允许将数据广播到多个进程，但由于每条消息都发送到每个订阅服务器，因此无法扩展处理。
我觉得这对我来说很有意义。
但是对于Kafka来说。文档说，即使在发布-订阅模式下，Kafka也可以扩展消息处理。(如果我错了，请纠正我。谢谢。)
卡夫卡的消费群体概念概括了这两个概念。与队列一样，使用者组允许您将处理划分为进程的集合(使用者组的成员)。与发布订阅一样，Kafka允许您向多个消费群体广播消息。
卡夫卡模型的优点在于，每个主
浏览 1提问于2017-07-16得票数 0
我正在尝试从最早的偏移量中读取卡夫卡主题，然后通过python脚本对某些记录进行墓碑。由于消息数量巨大(百万+)，我希望利用多处理来使脚本在使用消息时更快。下面是脚本中的一段代码：
from kafka import KafkaConsumer
def cleanup_kafka_topic(self, env):
# Declarations
consumer = KafkaConsumer(<topic_name>, group_id=<some_group>),
boots
浏览 71提问于2021-02-19得票数 0
我们使用Kafka来存储由集群中的节点生成的消息，并分发到集群中的所有节点，我让它主要使用akka-streams，但有几个问题我必须解决。这是有一些限制的。
首先，消息必须由集群中的每个节点使用，但只由一个节点生成。我知道我可以为每个节点分配一个组id，这可能是它的节点ID，这意味着每个节点都会收到消息。已经排序好了。但这里有一些问题。
数据是非常短暂和相当大的(仅低于一兆字节)，不能进一步压缩或分解。如果有关于这个主题的新消息，那么旧消息几乎是垃圾。如何才能将主题限制为当前最多只有一条消息？
考虑到数据是节点启动所必需的，我需要消费关于该主题的最新消息，无论我以前是否消费过它，并且希望不会
浏览 0提问于2017-12-06得票数 0
我们使用Storm开发了一个摄取应用程序，它使用Kafka消息(一些时间序列传感器数据)并将这些消息保存到Cassandra中。我们使用Nifi工作流程来实现这一点。
我现在要开发一个单独的星火流应用程序，它需要消费这些Kafka消息作为源。我想知道为什么当两个应用程序与一个Kafka Chanel交互时会出现问题？我是否应该将Nifi中的Kafka消息复制到另一个Chanel，以便我的Spark Streaming应用程序使用它们，但这是一个开销。
浏览 0提问于2017-04-30得票数 0
Server提供功能。您可以看到该文件从例如csv文件中读取并插入到表中。
我的理解是，这在与Kafka合作时有一个明显的缺点：
你必须把卡夫卡的信息转换成CSV
您必须接受kafka消息，在上一步转换之后，将其写入磁盘，以便大容量插入可以访问该文件。
我的问题是如何克服上述缺点；整个过程中的一些东西看起来是错误的。最让我担心的是第二个缺点，写到磁盘上。我是否能够将文件写入内存，然后对其执行大容量插入？
浏览 4提问于2021-10-22得票数 0
描述
我一直在尝试测试我发送给Kafka的一大块数据的正确性。当我试图在fabric中使用多重处理时，我混淆了这个过程以及消息使用者。消息使用者一开始没有被正确关闭，然后它就停止使用消息了。
之后，我在本地的机器上重新启动了Kafka (我用的是码头，所以我用了
docker-compose -f docker-compose-single-broker.yml rm
为了删除我用来测试的kafka，并使用以下方法重新创建一个新的
docker-compose -f dokcer-compose-single-broker.yml up
在kafka和kafka-manager启动并运行之后，
浏览 0提问于2018-09-25得票数 1
1回答
Camel JMS组件多线程
、、、、
我有一个spring boot应用程序，使用Camel的jmsComponent处理来自多个Solace队列的消息。我为每个队列连接定义了一个Camel route，并定义了一个将消息转换为json的Camel Processor。有大量来自队列的消息，而我的程序无法跟上速度。 路由： @Component
public class SomeRoute extends RouteBuilder {
@Override
public void configure() throws Exception {
from("jms:{{queue.name}}&
浏览 22提问于2020-07-21得票数 1
注意到一个问题，其中Kafka消费者组(用java实现)始终遗漏了来自代理的几条消息。作为调试的第一行，通过kafka控制台使用者，我可以在代理中看到这些消息。
Kafka broker版本: 0.10.1.0
Kafka客户端版本：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.9.0.1</version>
</dependency>
卡夫
浏览 2提问于2017-12-22得票数 6
JSON消息正在由字符串使用者使用。我的produces发送两种类型的消息字符串和序列化的JSON
@KafkaListener(topics = "${kafka.topic}", groupId = "${kafka.group-id}", containerFactory = "${kafka.string-listener-container-factory}")
public void consume(@NotNull ConsumerRecord<String, String> cr, @Payload String pa
浏览 0提问于2020-02-13得票数 1
2回答
我有一个关于管理多个CG的问题，创建了三个消费者组，每个CG都有自己的kafka服务、组Id和主题。
现在我收到了预期的消息，但是，我想知道是否有可能创建下一个场景：
创建三个消费者组，但只接收来自一个用户的消息，暂时暂停/暂停其他用户组，如果他的kafka服务将下降，则使用下一个消费者组的消息，与第三个用户组相同。
下面是我的代码示例：
function createConsumerGroup(topics){
const ConsumerGroup = kafka.ConsumerGroup;
//CREATE CONSUMER GROUPS FOR EVERY SER
浏览 1提问于2018-07-12得票数 2
回答已采纳
如果在整个网络的某个地方有一个Kafka服务器“在那里”，我会假设消费者可能有两种方式可以使用这些消息：
首先，通过“订阅”主题，并实际上告诉Kafka服务器它正在监听的位置，这样当产生新消息时，Kafka就会主动地通过网络将消息发送给消费者。
消费者必须投票卡夫卡服务器，要求任何新的消息，使用偏移的信息，它目前已采取。
这就是卡夫卡的工作方式，是否可以为哪一种选择配置？
浏览 3提问于2016-03-07得票数 5
1回答
在kafka使用者中，如果消息的处理时间超过5分钟，则消息是新处理的，我已经将使用者配置为增加"max.poll.interval.ms“和session.timeout.ms。
max.poll.interval.ms= 7200000 (2小时) session.timeout.ms= 7200000 (2小时) request.timeout.ms=7206000 (~2小时)
这开始起作用了，在那之后，它是重新平衡的，我在消费者身上得到了这个错误。
2019-10-18 15:29:51.739 INFO o.a.k.c.c.i.AbstractCoordinator -
浏览 3提问于2019-10-18得票数 4
我在spring boot应用程序的消费者接收器上使用@SendTo注释，没有参数。收到的消息包含头部"Kafka_replyTopic"，但@SendTo无法读取，kafka在发送回复时抛出异常： Listener failed; nested exception is java.lang.IllegalStateException: With no topic header, a defaultTopic is required
浏览 64提问于2020-09-02得票数 0
回答已采纳
使用该配置
使用spring-integration-kafka扩展和以下配置：
<int-kafka:zookeeper-connect id="zookeeperConnect"
zk-connect="#{kafkaConfig['zooKeeperUrl']}" zk-connection-timeout="10000"
zk-session-timeout="10000" zk-sync-time="2000" />
<int-kafka:consume
浏览 0提问于2016-01-29得票数 0
我有一个spring boot应用程序。我希望实现一个从Kafka主题消费的函数，对于每个消费的消息，它调用另一个方法对消息进行一些处理，然后将其存储在数据库中。源kafka主题有12个分区。我想在同一个消费群体下创建6到12个kafka消费者。所有的消费者都将消费来自Kafka的消息并调用该函数。我不能得到一个好的设计。有人能帮帮我吗？
浏览 1提问于2020-02-19得票数 1
我正在使用Kafka构建一个数据处理管道。流水线是线性的，有4级。数据量是中等的(将需要超过一台机器，但不需要数百或数千；数据量是几十or )我的问题是:我只能使用Kafka，让流水线阶段从一个主题消费并在另一个主题上生产吗？我应该使用Spark还是Storm ?为什么？当然，我更喜欢尽可能简单的架构。如果我可以用Kafka做所有的事情，我会更喜欢。在未来，我可能需要一些额外的机器学习阶段，这可能会影响答案。我没有很强的只有一次的语义，我可以接受一些消息丢失和一些重复，没有问题。
浏览 3提问于2015-04-01得票数 0
1回答
当我尝试使用Kerberized集群中的Spark流应用程序使用Kafka的消息时，我遇到了一些问题。我尝试了两种方法-- ：
基于接收机的方法：KafkaUtils.createStream
直接进场(无接收器)：KafkaUtils.createDirectStream
基于接收器的方法(KafkaUtils.createStream)引发两种类型的异常(不同的异常，无论我处于本地模式(--master local[*])还是纱线模式(--master yarn --deploy-mode client) )：
星火本地应用程序中奇怪的kafka.common.Brok
浏览 9提问于2017-12-26得票数 4
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
