如何将消息从一台PC上的Kafka制作人发送到另一台PC上的Kafka经纪人？

Question

我试图用wifi接口从一台电脑上的Kafka制作人发送消息到另一台电脑上的Kafka经纪人，但消息不出现在Kafka经纪人的指定主题中。

我用华硕无线路由器连接了两台PC机，并禁用了PC机和路由器上的所有防火墙。两台PC都成功地ping了对方。当我转向有线连接时，它工作了，消息被摄取到kafka broker PC上的指定主题。

Kafka制片人：

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

        CarDataProducer fProducer= new CarDataProducer();
        Producer<String, CarData> producer= fProducer.initializeKafkaProducer();

        String topicName = "IN-DATA";

        CSVReaderCarData csvReader = new CSVReaderCarData();
        List<CarData> CarDataList = csvReader.readCarDataFromCSV("data/mllib/TrainTest_101.csv");

        //read from CSV file and send
        for (CarData val : CarDataList) {
            producer.send(new ProducerRecord<String, CarData>(topicName, val));
        }
    }

    public KafkaProducer<String, CarData> initializeKafkaProducer() {

        // Set the producer configuration properties.
        Properties props = ProducerProperties.getInstance();

        // Instantiate a producerSampleJDBC
        KafkaProducer<String, CarData> producer = new KafkaProducer<String, CarData>(props);

        return producer;
    }

public class ProducerProperties {
    private ProducerProperties() {
    }

    public static final Properties props = new Properties();
    static {
        props.put("bootstrap.servers", "192.168.1.124:9092");
        props.put("acks", "0");
        props.put("retries", 0);
        props.put("batch.size", 500);
        props.put("linger.ms", 500);
        props.put("buffer.memory", 500);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "com.iov.safety.vehicleproducer.CarDataSerializer");
    }

    public static Properties getInstance() {
        return props;
    }
}

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic IN-DATA

 kafka-console-consumer.sh --bootstrap-server 192.168.1.124:9092 --topic IN-DATA
{"instSpeed":19.0,"time":15.0,"label":0.0}
{"instSpeed":64.0,"time":15.0,"label":1.0}
{"instSpeed":10.0,"time":16.0,"label":0.0}

null

Listeners=明文：//：9092

• netstat-anogrep'9092'

tcp6 0 0:::9092::*
监听(0.00/0/0)tcp6 0 0 127.0.0.1:53880
127.0.1.1:9092建立keepalive(6659.53/0/0)tcp6 0 0 127.0.1.1:9092建立keepalive(6659.53/0/0)tcp6 0 0 127.0.1.1:9092建立keepalive(6659.15/0/0)tcp6 0 0 127.0.1.1:53878建立keepalive(6659.15/0/0)tcp6 0 0 127.0.0.1:53878 127.0.1.1:53878 127.0.1.1:9092建立keepalive(6659.15/0/0)

通过向kafka生产者的发送添加回调，我得到超时错误：

timeoutexception：自上次追加后，in-data-0:30045毫秒的8条记录过期

Answer 1

我解决了。每个Kafka经纪人必须公布其主机名/IP，以便从另一台PC上的Kafka制作人那里获得。

 kafka-server-start.sh config/server.properties --override  advertised.listeners=PLAINTEXT://192.168.1.124:9092
sha

相反，我们可以按以下方式更新config/server.properties：

advertised.listeners=PLAINTEXT://your.host.name:9092 

或

advertised.listeners=PLAINTEXT://192.168.1.124:9092