一、環境準備：

1、maven工程中引入依賴：

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.12</artifactId>
        <version>1.1.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>1.1.0</version>
    </dependency>
 

public class KafkaProperties {
    //blocker連線
    public static final String broker = "192.168.80.140:9092";
    //topic
    public static final String topic = "kafka_api";
}

hostname和埠是用來建議給生產者和消費者使用的，如果沒有設定，將會使用listeners的配置，如果listeners也沒有配置，將使用java.net.InetAddress.getCanonicalHostName()來獲取這個hostname和port，基本就是localhost。
"PLAINTEXT"表示協議，可選的值有PLAINTEXT和SSL，hostname可以指定IP地址，也可以用"0.0.0.0"表示對所有的網路介面有效，如果hostname為空表示只對預設的網路介面有效
如果你沒有配置advertised.listeners，就使用listeners的配置通告給訊息的生產者和消費者，這個過程是在生產者和消費者獲取源資料(metadata)。
 

二、java API

1、實現生產者（Producer）

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class MyProducer extends Thread {
    private String topic;
    private KafkaProducer<String,String> producer;

    public MyProducer(String topic) {
        this.topic = topic;

        Properties properties = new Properties();
        properties.put("bootstrap.servers",KafkaProperties.broker);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<String, String>(properties);
    }

    @Override
    public void run() {
        int messageNo = 1;
        while (true){
            String message = "message" + messageNo;
            System.out.println("send = "+message);
            producer.send(new ProducerRecord<String, String>(topic,message));
            messageNo ++;

            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args){
        new MyProducer(KafkaProperties.topic).start();
    }
}
 

生產者控制檯輸出結果：

2、實現消費者（Consumer）

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;

public class MyConsumer extends Thread {
    private String topic;
    KafkaConsumer<String, String> consumer;

    public MyConsumer() {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", KafkaProperties.broker);
        properties.setProperty("group.id","test");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<String, String>(properties);
        consumer.subscribe(Arrays.asList(KafkaProperties.topic));
    }

    @Override
    public void run() {
        while (true){
            ConsumerRecords<String,String> records = consumer.poll(100);
            for (ConsumerRecord<String,String> recode: records) {
                System.out.println("recodeOffset = " + recode.offset() + "recodeValue = " + recode.value());
            }
        }
    }
    public static void main(String[] args){
        new MyConsumer().start();
    }
}

消費者控制檯輸出結果：