KafkaConsumerAPI - 从eclipse连接到vm中通过vagrant运行的代理

Question

什么是advertised.listeners。它如何帮助连接到从Eclipse运行的VM中的Kafka代理？

我编写了KafkaConsumer代码并使用本地运行的服务对其进行了测试。我正在尝试使用相同的代码连接到在VM中运行的Kafka Broker，但记录不会被消耗掉。请解释实现结果所需的配置。

package kafkaExample;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;

public class Consumer {
  public static void main(String[] args) {

    final List<PartitionInfo> partitionInfos;


    final Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, args[0]);
    /*    props.put("zookeeper.connect", "10.30.3.2:2181,10.30.3.3:2181");*/
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "cgrp1");
    props.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer-tutorial");


    final KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
    kafkaConsumer.subscribe(Arrays.asList(args[1]),
        new ConsumerRebalanceListener() {
          @Override
          public void onPartitionsRevoked(Collection<TopicPartition> arg0) {

        }

          @Override
          public void onPartitionsAssigned(Collection<TopicPartition> tps) {

        }
        });

    System.out.println("list of topics " + kafkaConsumer.listTopics());

    ConsumerRecords<String, String> records = null;
    while (true) {
      records = kafkaConsumer.poll(Duration.ofMillis(100));
      if (records == null || records.count() == 0) {
        System.out.println("There is no message. try again.");
        continue;
      }

      final Iterator<ConsumerRecord<String, String>> iter = records.iterator();
      while (iter.hasNext()) {
        final ConsumerRecord<String, String> cr = iter.next();
        System.out.println("Consume msg: " + cr.value());
      }
      kafkaConsumer.commitSync();
      kafkaConsumer.close();
      break;
    }
  }
}

Answer 1

广告的侦听器将告诉您的客户端连接到与服务器本身的主要地址不同的地址。

例如，您的VM至少有三个接口，localhost，用于跨多个VM进行通信的内部VM子网，然后可能还有另一个用于外部主机进入该VM的接口。

第三个不能存在，如果没有，那么您需要设置从主机到VM内部接口的端口转发。在从主机9092（可以是任何数字）转发到Kafka服务器9092的端口的情况下，然后您将广告的侦听器设置为PLAINTEXT://127.0.0.1:9092（或从主机中选择的号码）并且您将使您申请主机然后连接到127.0.0.1:9092，而不是VM地址

如果您在一台主机上有一组VM，则在转发时必须使用不同的端口，例如127.0.0.1:9092,127.0.0.1:9093