Spring-Kafka并发性

Question

我正在用Spring-Kafka写我的第一个Kafka消费者。看了一下framework提供的不同选项，对相同的选项几乎没有疑问。能否有人请澄清以下，如果你已经在它工作。

问题1：根据Spring-Kafka文档，有两种实现Kafka-Consumer的方法；“您可以通过配置MessageListenerContainer并提供消息侦听器或使用@Kafkalistener注释来接收消息”。有人能告诉我什么时候应该选择一个选项而不是另一个选项吗？

问题2：我选择了KafkaListener方法来写我的申请。为此，我需要初始化一个容器工厂实例，在容器工厂内部有控制并发的选项。我只想检查一下我对并发性的理解是否正确。

批处理确认使用者：

    public void onMessage(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment,
          Consumer<?, ?> consumer) {
      for (ConsumerRecord<String, String> record : records) {
          System.out.println("Record : " + record.value());
          // Process the message here..
          listener.addOffset(record.topic(), record.partition(), record.offset());
       }
       acknowledgment.acknowledge();
    }

初始化集装箱工厂：

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<String, String>(consumerConfigs());
}

@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> configs = new HashMap<String, Object>();
    configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServer);
    configs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enablAutoCommit);
    configs.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPolInterval);
    configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
    configs.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
    configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return configs;
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
    // Not sure about the impact of this property, so going with 1
    factory.setConcurrency(2);
    factory.setBatchListener(true);
    factory.getContainerProperties().setAckMode(AckMode.MANUAL);
    factory.getContainerProperties().setConsumerRebalanceListener(RebalanceListener.getInstance());
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setMessageListener(new BatchAckConsumer());
    return factory;
}

Answer 1

>

@kafkalistener是一个消息驱动的“POJO”，它添加了负载转换、参数匹配等内容。如果实现MessageListener，则只能从Kafka获得原始的ConsumerRecord。参见@Kafkalistener注释。

是的，并发表示线程数；每个线程创建一个使用者；它们并行运行；在您的示例中，每个分区将获得2个分区。

如果我们是平行消费，我们是否应该考虑任何事情。