监听多个分区的一个KafkaConsumer与监听多个分区的多个KafkaConsumers
问题：监听多个分区的一个KafkaConsumer与监听多个分区的多个KafkaConsumers有什么区别？
回答：
监听多个分区的一个KafkaConsumer与监听多个分区的多个KafkaConsumers在实现上有一些区别。
监听多个分区的一个KafkaConsumer：
- 概念：一个KafkaConsumer实例可以同时监听多个分区。
- 分类：属于单个KafkaConsumer的工作模式。
- 优势：通过一个KafkaConsumer实例监听多个分区可以减少资源消耗和管理复杂性。
- 应用场景：适用于分区数量相对较少且能够被单个KafkaConsumer处理的情况。
- 推荐的腾讯云相关产品：腾讯云消息队列 CMQ（https://cloud.tencent.com/product/cmq）
监听多个分区的多个KafkaConsumers：
- 概念：多个KafkaConsumer实例同时监听多个分区。
- 分类：属于多个KafkaConsumer协同工作的模式。
- 优势：多个KafkaConsumer实例可以并行处理消息，提高消费能力和吞吐量；同时也增加了系统的弹性和可靠性。
- 应用场景：适用于分区数量较多，或者需要更高的消费能力和吞吐量的场景。
- 推荐的腾讯云相关产品：腾讯云消息队列 CMQ（https://cloud.tencent.com/product/cmq）、腾讯云Serverless Cloud Function（SCF）（https://cloud.tencent.com/product/scf）
需要注意的是，以上仅为对监听多个分区的一个KafkaConsumer与监听多个分区的多个KafkaConsumers的概念、分类、优势、应用场景以及推荐的腾讯云相关产品的介绍，并没有涉及具体的编程语言或技术实现。具体的实现方式和使用方法可能因所选编程语言、技术框架、云平台等而有所不同。
相关·内容
Kafka Connect中的任何选项都可以指定从哪个分区读取消息。基本上，我正在寻找一个选项，在Kafka连接中手动分配要读取的分区列表。 类似于KafkaConsumer API中的assign()方法 https://kafka.apache.org/0100/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)
浏览 9提问于2019-03-29得票数 1
回答已采纳
1回答
kafka版本: 0.9.0.1
如果是n = 20，我必须获取一个主题的最后20条消息。
我试着用
kafkaConsumer.seekToBeginning();
但它会检索所有的消息。我只需要收到最后20条消息。
此主题可能有数十万条记录
public List<JSONObject> consumeMessages(String kafkaTopicName) {
KafkaConsumer<String, String> kafkaConsumer = null;
boolean flag = true;
List<JSONObject>
浏览 0提问于2019-04-18得票数 3
回答已采纳
2回答
我正在尝试使用assign()方法消费来自多个主题的消息。通过我的实现，有时我能够使用来自所有主题的消息，而另一些时候只能使用一个主题的消息。经过研究，我发现Kafka默认使用范围赋值器。因此，它不会总是分配所有分区。
对于我的用例，我应该能够使用所有主题和分区。
我已经尝试通过设置RoundRobin分配器。但这并没有帮助
List<TopicPartition> topicPartitions = new ArrayList<>();
KafkaConsumer kafkaConsumer = new KafkaConsumer<>(consumerCo
浏览 0提问于2019-06-17得票数 0
1回答
我有一个具有多个主题的Kafka-Broker，每个主题都有一个分区。
我有一个消费者，它可以很好地使用主题中的消息
我的问题是，我需要通过增加分区的数量来提高消息队列的吞吐量，假设我在一个主题上有四个分区，是否有一种方法可以编写四个消费者，每个消费者指向该主题上的单个分区？
import java.util.*;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStre
浏览 25提问于2016-09-23得票数 4
回答已采纳
所以，我有一个@Component课程来听卡夫卡的主题
@Component
@Data
@Slf4j
public class KafkaConsumer {
public List<String> saveReserveStock = new ArrayList<>();
@KafkaListener(topics = "topic")
public void listenReserveStock(ConsumerRecord<?, ?> consumerRecord) {
System.out.
浏览 1提问于2022-07-10得票数 0
1回答
我一直在使用python模块来使用kafka代理。我想使用相同的主题并行地使用'x‘数量的分区。这些文件的内容如下：
# Use multiple consumers in parallel w/ 0.9 kafka brokers
# typically you would run each on a different server / process / CPU
consumer1 = KafkaConsumer('my-topic',
group_id='my-group',
浏览 2提问于2016-05-24得票数 8
回答已采纳
2回答
我已经看到了一个类似于的问题
但是，我仍然想知道来自特定分区的流数据是否不可能实现？我使用了卡夫卡消费策略在星火流订阅方法。
ConsumerStrategies.Subscribe
这是我尝试订阅主题和分区的代码片段，
val topics = Array("cdc-classic")
val topic="cdc-classic"
val partition=2;
val offsets=
Map(new TopicPartition(topic, partition) -> 2L)//I am not clear with this lin
浏览 0提问于2018-06-07得票数 0
回答已采纳
我使用下面的代码来读取Kafka主题，并处理数据。
JavaDStream<Row> transformedMessages = messages.flatMap(record -> processData(record))
.transform(new Function<JavaRDD<Row>, JavaRDD<Row>>() {
//JavaRDD<Row> records = ss.emptyDataFrame().toJavaRDD();
浏览 1提问于2017-06-14得票数 11
我正在尝试从最早的偏移量中读取卡夫卡主题，然后通过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版本2.0和java使用者API来使用来自主题的消息。我们使用的是一个单节点Kafka服务器，每个分区只有一个使用者。我注意到消费者正在失去一些信息。场景是:消费者投票这个话题。我已经创建了一个消费者每线程。获取消息并将其交给处理程序来处理消息。然后用“至少一次”的卡夫卡消费者语义学来提交抵消，以实现卡夫卡的偏移。同时，我有另一个使用者使用不同的组-id运行。在这个使用者中，我只是增加消息计数器并提交偏移量。在这个消费者身上没有信息丢失。
try {
//kafkaConsumer.registerTopic();
consumerThread = new
浏览 1提问于2019-05-10得票数 1
1回答
我试图让x数的消费者访问kafka中指定的主题，但不使用相同的消息。比如我想要..。
消费者1拾取偏移1消费者2拾取偏移2消费者1拾取偏移3消费者2拾取偏移4
我想让卡夫卡为这两位消费者排队。我注意到了group.id配置，我假设您可以使用同一个组，它会相应地处理它，但它似乎不像我想的那样工作。
这是我使用的代码..。
public void init(){
Properties props = new Properties();
props.put("bootstrap.servers", kafkaUrl);
浏览 2提问于2016-12-06得票数 0
回答已采纳
我使用的是KafkaConsumer 0.10Java api。我想从一个特定的分区和特定的偏移量消费。我抬头一看，发现有一个seek方法，但它抛出了一个异常。有没有人有类似的用例或解决方案？
代码：
KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(consumerProps);
consumer.seek(new TopicPartition("mytopic", 1), 4);
异常
java.lang.IllegalStateException: No current ass
浏览 0提问于2016-12-07得票数 21
回答已采纳
当使用相同的topic和相同的group-id时，我希望能够使用来自两个不同消费者的数据(每个消费者将得到不同的消息)。
制片人：
producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
value_serializer=lambda x:
dumps(x).encode('utf-8'))
for e in range(50):
data = {'
浏览 4提问于2020-10-19得票数 2
回答已采纳
1回答
我已经使用python.How创建了实现的Kafka生产者-消费者消息传递主题我可以对队列做同样的事情，这样消息就只会发送给一个消费者。
这是我的生产者代码
# Import KafkaProducer from Kafka library
from kafka import KafkaProducer
# Define server with port
bootstrap_servers = ['localhost:9092']
# Define topic name where the message will publish
topicName = 'Firs
浏览 3提问于2020-12-03得票数 0
我是卡夫卡的新手，我通过了，但我什么都不懂。有人能解释一下什么时候使用ConcurrentKafkaListenerContainerFactory类吗？我已经使用了Kafkaconsumer类，但是我发现在我当前的项目中正在使用ConcurrentKafkaListenerContainerFactory。请解释一下它的用途。
浏览 2提问于2019-03-06得票数 11
回答已采纳
1回答
我有一个设置，其中每个KafkaConsumers在一个主题上处理多个分区。它们被静态地分配分区，以确保每个使用者都有相同数量的分区要处理。还选择了记录键，以便在所有分区上具有相同的消息分布。
在负载较重的时候，我们经常会看到少数分区会产生相当大的延迟(数千条消息/几分钟值)，而其他正在获得相同负载且被同一个使用者占用的分区则设法将延迟降低到几百条消息/几秒钟。
看起来消费者正在尽可能快地获取记录，遍历大部分分区，但有时会有一个分区被遗漏很长时间。理想情况下，我希望看到延迟更均匀地分布在各个分区上。
我已经阅读了一段时间关于KafkaConsumer民意测验行为和配置的文章，到目前为止，我认为
浏览 5提问于2019-12-27得票数 5
1回答
春天-卡夫卡听众同意
、、、
我使用实现了Kafka消费者。我有一个带有2个分区的Kafka主题，并且我使用了并发级别设置为2的ConcurrentKafkaListenerContainerFactory，因此每个容器实例都应该按照spring 从单个分区中使用。
KafkaMessageListenerContainer在一个线程上接收来自所有主题/分区的所有消息。ConcurrentMessageListenerContainer委托一个或多个KafkaMessageListenerContainer来提供多线程消耗。
这是我的消费类：
@Component
public class KafkaConsume
浏览 3提问于2017-09-21得票数 5
回答已采纳
用例：我们有为多模式配置的金门卡夫卡大数据适配器。每个模式映射到一个分区，kafka分区器按照定义的映射发布消息。比如schema1=1(Partition#1)，schema2=2 .scheman=n
这样，所有模式的消息都隔离在它们自己的分区中。
问题:给定情况下的，我想配置一个Apachi Nifi节点实例，以处理kafka主题的1个分区。这样，我将配置运行在并行处理100个kafka分区中的100个Nifi实例。
但我看不到在KafkaConsumer中可以指定分区的任何地方
问题：
Apache使用者支持分区配置吗？
如果它不支持，它是否在未来发布的路线图中？
浏览 2提问于2017-11-10得票数 1
回答已采纳
我有一个有两个分区的kafka主题，我想创建一个使用者从给定的偏移量读取主题，下面是我用来从给定偏移量(9)读取的示例代码。
Properties configProperties = new Properties();
configProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
configProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.seri
浏览 0提问于2019-06-20得票数 1
回答已采纳
我有一个小问题，那就是我必须使用kafka作为通知系统来启动我的应用程序流，一切都很好，但是我想知道，当队列中有未读消息时，我是否可以改变顺序。
让我解释一下:当主题中有两条或两条以上的信息，而消费者读取它们时，先读，最后一条输入，我希望它先读，第一条输入。
所以我想知道这是否可能，如果可能的话，我们必须为此而改变。
我正在使用Spring的最新版本，和Kafka 2.5.0
ConsumerConfig：
@EnableKafka
@Configuration
public class KafkaConsumerConfig {
private static final Strin
浏览 0提问于2020-06-18得票数 3
回答已采纳
def getKafkaConsumer(topicName, kafkaIP, kafkaPort, kafkaConsumerGroup): try: bootstrap = [str(kafkaIP + ':' + kafkaPort)] consumer = KafkaConsumer(topicName, bootstrap_servers = bootstrap, group_id = kafkaConsumerGroup, session_timeout_ms = 30000, api_version = (0,10)) return consumer excep
浏览 10提问于2022-08-20得票数 0
我在Kubernetes上运行Kafka Connect (8-16个节点，自动标度)。我总共定义了44个连接器，每个Kafka主题一个(每个主题一个分区)。这些主题是由Debezium / Postgresql编写的。有3个Kafka节点。每个连接器的tasks.max设置为4。大部分连接器(但不是每个连接器！)有一个(总是一个)失败的任务，由于java.lang.IllegalStateException:没有当前分配的分区-0。
这里不是卡夫卡专家，请注意;)我假设有3个Kafka节点，因此3个工作人员做得很好，而第四个任务没有任何连接，所以失败了。但是为什么有时有4个任务运行得很好呢？
浏览 3提问于2019-03-24得票数 0
回答已采纳
StreamsBuilder builder = new StreamsBuilder();
Map<String, ?> serdeConfig = Collections.singletonMap(SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
Serde keySerde= getSerde(keyClass);
keySerde.configure(serdeConfig,true);
Serde valueSerde = getSerde(valueClass);
valueSe
浏览 0提问于2019-07-25得票数 2
回答已采纳
我一直试图使用AdminClient来获取消费者的滞后，但是adminClient.listGroupOffsets("foo")；返回一个空指针NullPointerException。这是我的代码：
public long getLag() {
AdminClient adminClient = AdminClient.createSimplePlaintext("localhost:9092");
scala.collection.immutable.Map<TopicPartition, Object> offsets = a
浏览 3提问于2017-06-27得票数 0
回答已采纳
我创建了一个名为ConsumerConfig和Service的类，它包含从主题获取记录的函数。
consumer.poll只从一个分区消费，我想我需要添加一个循环，但在本例中(True)并不是最好的选择
调试时，我计算这个函数"records = consumer.poll(Duration.ofMillis(4000))“，每次计算它时，它都从下一个分区读取。
下面是ConsumerConfig类的代码和从特定主题获取记录的函数：
@Component
public class ConsumerConfig {
@Autowired
KafkaProperties
浏览 1提问于2022-06-15得票数 0
1回答
我创建了一个包含3个分区的主题
Topic:fixtures PartitionCount:3 ReplicationFactor:1 Configs:
Topic: fixtures Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Topic: fixtures Partition: 1 Leader: 0 Replicas: 0 Isr: 0
Topic: fixtures Partition: 2 Leader: 0 Replicas: 0 Isr: 0
我使用Java Producer同步
浏览 4提问于2018-02-27得票数 1
回答已采纳
1回答
我有一个应用程序，它有两个线程正在运行。这些线程在它们之间共享一个KafkaConsumer实例。
KafkaConsumer是通过禁用自动提交配置创建的。一个线程将根据某些外部输入将主题分区分配给KafkaConsumer实例。另一个线程不断轮询主题，处理记录并在结束时提交偏移量。
现在，假设主题-分区分配在线程-1中发生了更改。我们如何确保Thread-2能够看到这些更改？也就是说，线程-2的KafkaConsumer将知道主题-分区已经在线程-1中更新。
将变量定义为易失性是否解决了这个问题？还是我们应该再做点什么？请提供您的解释与代码示例，因为我是新的多线程在Java。
浏览 4提问于2021-10-26得票数 2
回答已采纳
我试着从卡夫卡那里按分区消费。我正在使用KafkaConsumer api来完成这个任务。我看到这里提供了一个pause和resume方法，我可以从文档中收集到这是我想要的行为。
挂起从请求的分区获取。将来对轮询(Long)的调用将不会从这些分区返回任何记录，直到使用简历(Collection)恢复这些记录为止。请注意，此方法不影响分区订阅。特别是，当使用自动分配时，它不会导致组重新平衡。
我关心的是，我不会间歇地暂停和恢复KafkaConsumer，而是暂停所有的分区，除了我想要使用的分区。我担心暂停的调用代价高昂，会导致吞吐量问题。
我不确定这是否是实现来自KafkaConsume
浏览 0提问于2018-08-02得票数 1
我使用了下面的Kafka消费者示例代码：
KafkaConsumer kconsumer = new KafkaConsumer(props);
ConsumerRecords<String, byte[]> records = kconsumer.poll(timeoutInMS);
kconsumer.commitSync();
我有两个问题：
(1)偏移量提交的地方是Kafka还是Zookeer？(我知道旧的kafka版本是提交给zookeeper的，但是新的kafka版本仍然是ZK，我甚至没有在配置文件中指明zookeeper的IP )。
(2)如果需要重复使用数据，如何检
浏览 4提问于2018-01-12得票数 0
当源主题分区计数= 1时工作正常。如果我将分区增加到任何大于1的值，我会看到下面的错误。既适用于低级，也适用于DSL API。有什么建议吗？可能会遗漏什么？
org.apache.kafka.streams.errors.StreamsException: stream-thread [StreamThread-1] Failed to rebalance
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:410)
at org.apach
浏览 3提问于2017-02-19得票数 6
回答已采纳
以前我一直在使用0.8API。当您将主题列表传递给它时，它会返回一个流映射(每个主题一个条目)。这允许我生成一个单独的线程，并将每个主题的流分配给它。在每个主题中有太多的数据，产生一个单独的线程有助于多任务处理。
//0.8 code sample
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap =
consumer.createMessageStreams(topicCountMap);
我想升级到0.10。我检查了KafkaStreams和KafkaConsumer类。KafkaC
浏览 4提问于2016-09-30得票数 2
回答已采纳
1回答
我正在编写一个kafka节点用户，与python中的使用者相比，它非常慢，而且似乎没有接收到所有的消息。我试着调整所有可能的参数，但都没有效果。
我使用这个API：
这是我的密码：
//--------------------Create a consumer---------------------------------------------------------------------------
var kafka = require('kafka-node');
var Consumer = kafka.Consumer;
var client = new kaf
浏览 2提问于2018-05-07得票数 0
回答已采纳
我们希望在读取来自kafka的消息时实现并行性。因此，我们希望在flinkkafkaconsumer中指定分区编号。它将读取kafka中所有分区的消息，而不是特定的分区号。以下是示例代码： Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("zookeeper.connect", "localhost:2181");
prop
浏览 139提问于2020-07-09得票数 1
对于下面的代码，我使用了
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
consumer.subscribe(Arrays.asList(TOPIC));
TopicPartition partition = new TopicPartition(TOPIC, 0);
System.out.println(consumer.position(partition));
我收到类似“您只能检查分配给此使用者的分区的位置”之类的错误，上面的代码有什么
浏览 0提问于2017-09-24得票数 3
我正在查看KafkaConsumer源代码。从源代码(我不太熟悉的java中)来看，暂停一个主题分区似乎将其偏移状态设置为暂停。我正在试图弄清楚KafkaConsumer在后台所做的取取发生了什么，以及暂停主题分区是否停止了kafka使用者所做的预取，或者它是否继续预取，但它在轮询中不返回该主题分区的ConsumerRecords。
浏览 2提问于2018-10-22得票数 0
回答已采纳
1回答
我在这里询问的是关闭卡夫卡的消费者。即使使用线程退出，我也需要关闭kafka使用者吗？会不会因为任何改变而关闭它的资源？
下面是一个代码示例：
public class MyThread extends Thread{
private KafkaConsumer<String, Message> kafkaConsumer;
@Override
public void run() {
kafkaConsumer = initConsumer();
while(true){
kafkaconsume
浏览 0提问于2018-08-16得票数 3
回答已采纳
我已经创建了一个带有4个分区的简单生产者，现在我希望在一个消费者组中创建4个消费者来使用来自每个分区的数据。我怎么能这么做？
消费者代码
public class KafkaConsumer {
static List<String> list = new ArrayList<String>();
public static DataFrame reqFieldBOIDDf;
public static DataFrame df ;
static SparkConf conf = new SparkConf()
浏览 2提问于2017-10-20得票数 0
1回答
当恢复所有主题的分区时，我不断地收到调试消息。如下所示。这条消息在我的服务器上每毫秒连续打印一次。
08:44:34.850 [default-akka.kafka.default-dispatcher-10] DEBUG o.a.k.clients.consumer.KafkaConsumer - Resuming partition test222-7
08:44:34.850 [default-akka.kafka.default-dispatcher-10] DEBUG o.a.k.clients.consumer.KafkaConsumer - Resuming partition
浏览 0提问于2016-12-07得票数 5
如何将python使用者脚本附加到特定的kafka分区。
在运行两个使用者脚本实例(如下所示)时，每个实例都随机获取一个分区，然后按照预期使用/打印该特定分区的所有消息。
但是，由于我需要将这些消息输出到磁盘上名为本地文件的分区，将脚本的每个实例附加到预先声明的分区ID将使事情变得更容易
档案名称(如。：
Date/Hour/PARTITION_ID-0.CSV
Date/Hour/PARTITION_ID-1.CSV
任何关于如何实现这一目标的想法。
请随意建议其他方法。
卡夫卡设置：
Topic:my-topic3 PartitionCount:2 ReplicationFactor
浏览 1提问于2016-06-25得票数 2
回答已采纳
2回答
要多次订阅给定的Kafka主题并从每个KafkaConsumer的每个分区接收每条消息，规范的方法是什么？
我目前正在做的是生成一个随机的Uuid group.id，这样每个订阅都是一个新的组，但是考虑到这些订阅是短暂的(并且有很多)，Kafka存储关于它们的元数据的开销可能是有害的。
实现这一点的正确方法是什么？
浏览 1提问于2020-09-07得票数 0
2回答
我想为几个主题创建一个kafka消费者。consumer的方法构造函数允许我在订阅中传递主题列表的参数，如下所示：
private Consumer createConsumer() {
Properties props = getConsumerProps();
Consumer<String, byte[]> consumer = new KafkaConsumer<>(props);
ArrayList<String> topicMISL = new ArrayList<>();
for (String s:
浏览 0提问于2020-06-22得票数 2
我有下面的程序来消费所有来卡夫卡的消息。
from kafka import KafkaConsumer
consumer = KafkaConsumer('my_test_topic',
group_id='my-group',
bootstrap_servers=['my_kafka:9092'])
for message in consumer:
consumer.commit()
print ("%s key=%s
浏览 4提问于2017-07-17得票数 6
回答已采纳
2回答
我无法理解这个API的设计！
在下面的代码中，我们订阅了一个具有动态分配分区的主题列表。这完全没问题。
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
consumer.subscribe(Arrays.asList("some-topic"));
while(true){
ConsumerRecords<String, String> records = consumer
浏览 0提问于2019-07-27得票数 2
我使用的是Kafka 0.9.0.1。
我第一次启动应用程序时，需要20-30秒才能检索到主题中的“最新”消息
我使用过不同的Kafka代理(具有不同的配置)，但我仍然看到这种行为。对于后续的消息，通常不会出现速度缓慢的情况。
这是预期的行为吗？通过运行此示例应用程序并将代理/主题名称更改为您自己的设置，您可以在下面清楚地看到这一点
public class KafkaProducerConsumerTest {
public static final String KAFKA_BROKERS = "...";
public static final Stri
浏览 0提问于2016-07-23得票数 2
2回答
我正在尝试使用EventTime设置一个流解决方案。我将从卡夫卡那里消费。我的数据可能被延迟一分钟(所以我的WaterMark被延迟了一分钟)，并且出现了故障。我有30秒的窗户。我有以下设置：
KafkaConsumer.scala
object KafkaConsumer {
def main(args: Array[String]) {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventT
浏览 0提问于2018-06-14得票数 0
回答已采纳
1回答
我们正在努力实现卡夫卡作为我们的消息代理解决方案。我们正在IBM中部署Spring微服务，其内部message实现为BLuemix 0.10版本。由于我的经验更多地是关于JMS，ActiveMQ端，所以我想知道在java使用者中处理系统级错误的理想方法是什么？
以下是我们目前如何实现它
消费者特性
enable.auto.commit=false
auto.offset.reset=latest
我们使用的默认属性是
max.partition.fetch.bytes
session.timeout.ms
Kafka消费者
我们正在每个主题上旋转3个线程，每个主题都具有相同的groupId，即
浏览 1提问于2017-05-15得票数 1
我使用卡夫卡消费者在我们的应用程序与骆驼(骆驼-卡夫卡)。我们使用这些消息并将数据发送到服务器进行处理。有一个主题"XYZ“被定义为30个分区，我已经确定了15个作为每个消费者节点的使用者计数(总共2个实例)‘
CamelConsumer配置
kafka.consumersCount=15
kafka.consumerStreams=15
我从日志中看到，当使用者启动时，有15个使用者线程(假设在1个节点上)，这在配置上是很好的。
INFO Camel (camel-1) thread #2 - KafkaConsumer[XYZ] [camel.component.kafka.Ka
浏览 1提问于2019-09-24得票数 1
我在AWS的DC/OS (Mesos)集群上安装了Kafka。启用了三个代理，并创建了一个名为"topic1“的主题。
dcos kafka topic create topic1 --partitions 3 --replication 3
然后，我编写了一个Producer类来发送消息，一个Consumer类来接收消息。
public class Producer {
public static void sendMessage(String msg) throws InterruptedException, ExecutionException {
Map
浏览 1提问于2017-02-28得票数 8
回答已采纳
在consumer上获取此kafka异常：
org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'correlation_id': java.nio.BufferUnderflowException
at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:71)
at org.apache.kafka.common.requests.ResponseHeader.parse(ResponseHeader.
浏览 4提问于2017-06-07得票数 3
我正在尝试使用Apache Kafka中从主题开头开始的所有消息。我可以使用在那个时刻产生的消息。下面是我获取消息的代码。
public void consume() {
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(topic));
while (true) {
ConsumerRecords<String, String> records = consume
浏览 29提问于2018-07-25得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
