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

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 内部接口的端口转发。在这种情况下，从您的主机 9092（可以是任何数字）到 Kafka 服务器 9092 的端口转发，然后您将广告侦听器设置为 PLAINTEXT://127.0.0.1:9092（或从主机中选择的数字），然后您将然后主机上的应用程序连接到127.0.0.1:9092，而不是虚拟机地址

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