<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.11.0.0</version>
</dependency>

2、Producer

2.1、producer傳送訊息

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
/**
 * @author Thomas
 * @Description:最簡單的kafka producer
 * @date 22:18 2019-7-5
 */
public class ProducerDemo {
    public static void main(String[] args) {
        Properties properties =new Properties();
        //zookeeper伺服器叢集地址，用逗號隔開
        properties.put("bootstrap.servers", "172.16.0.218:9092,172.16.0.219:9092,172.16.0.217:9092");
        properties.put("acks", "all");
        properties.put("retries", 0);
        properties.put("batch.size", 16384);
        properties.put("linger.ms", 1);
        properties.put("buffer.memory", 33554432);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //自定義producer攔截器
        properties.put("interceptor.classes", "com.lt.kafka.producer.MyProducerInterceptor");
        //自定義訊息路由規則（訊息傳送到哪一個Partition中）
        //properties.put("partitioner.class", "com.lt.kafka.producer.MyPartition");

        Producer<string, string=""> producer = null;
        try {
            producer = new KafkaProducer<string, string="">(properties);
            for (int i = 20; i < 40; i++) {
                String msg = "This is Message:" + i;

                /**
                 * kafkaproducer中會同時呼叫自己的callback的onCompletion方法和producerIntercepter的onAcknowledgement方法。
                 * 關鍵原始碼：Callback interceptCallback = this.interceptors == null
                 * callback : new InterceptorCallback<>(callback,
                 * this.interceptors, tp);
                 */
                producer.send(new ProducerRecord<string, string="">("leixiang", msg),new MyCallback());
            }
        } catch (Exception e) {
            e.printStackTrace();

        } finally {
            if(producer!=null)
                producer.close();
        }
    }
}

2.2、自定義producer攔截器

import java.util.Map;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
 * @author Thomas
 * @Description:自定義producer攔截器
 * @date 22:21 2019-7-5
 */
public class MyProducerInterceptor implements ProducerInterceptor<string,string>{
    /**
     * 列印配置相關資訊
     */
    public void configure(Map<string,> configs) {
        // TODO Auto-generated method stub
        System.out.println(configs.toString());
    }

    /**
     * producer傳送資訊攔截方法
     */
    public ProducerRecord<string,string> onSend(ProducerRecord<string, string=""> record) {
        System.out.println("攔截處理前=============");
        String topic=record.topic();
        String value=record.value();
        System.out.println("攔截處理前的訊息====："+value);
        ProducerRecord<string,string> record2=new ProducerRecord<string, string="">(topic, value+" (intercepted)");
        System.out.println("攔截處理後的訊息："+record2.value());
        System.out.println("攔截處理後===============");
        return record2;
    }

    /**
     * 訊息確認回撥函式，和callback的onCompletion方法相似。
     * 在kafkaProducer中，如果都設定，兩者都會呼叫。
     */
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (metadata != null)
            System.out.println("MyProducerInterceptor onAcknowledgement:RecordMetadata=" + metadata.toString());
        if (exception != null)
            exception.printStackTrace();
    }
    /**
     * interceptor關閉回撥
     */
    public void close() {
        System.out.println("MyProducerInterceptor is closed!");
    }
}

2.3、自定義訊息路由規則

自定義路由規則，可以根據自己的需要定義訊息傳送到哪個分割槽。自定義路由規則需要實現Partitioner。

import java.util.Map;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
/**
 * @author Thomas
 * @Description:
 * @date 22:24 2019-7-5
 */
public class MyPartition implements Partitioner {
    public void configure(Map<string,> arg0) {
        // TODO Auto-generated method stub
    }

    public void close() {
        // TODO Auto-generated method stub
    }

    public int partition(String arg0, Object arg1, byte[] arg2, Object arg3, byte[] arg4, Cluster arg5) {
        // TODO Auto-generated method stub
        return 0;
    }
}

3、Consumer

3.1、自動提交

import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
/**
 * @author Thomas
 * @Description:
 * @date 22:26 2019-7-5
 */
public class AutoCommitConsumerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "172.16.0.218:9092,172.16.0.219:9092,172.16.0.217:9092");
        props.put("group.id", "leixiang");
        props.put("enable.auto.commit", "true");
        //想要讀取之前的資料，必須加上
        //props.put("auto.offset.reset", "earliest");
        /* 自動確認offset的時間間隔 */
        props.put("auto.commit.interval.ms", "1000");
        /*
         * 一旦consumer和kakfa叢集建立連線，
         * consumer會以心跳的方式來高速叢集自己還活著，
         * 如果session.timeout.ms 內心跳未到達伺服器，伺服器認為心跳丟失，會做rebalence
         */
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //配置自定義的攔截器，可以在攔截器中引入第三方外掛實現日誌記錄等功能。
        //props.put("interceptor.classes", "com.lt.kafka.consumer.MyConsumerInterceptor");

        @SuppressWarnings("resource")
        KafkaConsumer<string, string=""> consumer = new KafkaConsumer<string, string="">(props);
        try {
            /* 消費者訂閱的topic, 可同時訂閱多個 ，用逗號隔開*/
            consumer.subscribe(Arrays.asList("leixiang"));
            while (true) {
                //輪詢資料。如果緩衝區中沒有資料，輪詢等待的時間為毫秒。如果0，立即返回緩衝區中可用的任何記錄，則返回空
                ConsumerRecords<string, string=""> records = consumer.poll(100);
                for (ConsumerRecord<string, string=""> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(),
                        record.value());
            }
        } catch (Exception e) {
            // TODO: handle exception
            e.printStackTrace();
        }
    }
}

3.2、手動提交

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
/**
 * @author Thomas
 * @Description:
 * @date 22:28 2019-7-5
 */
public class ManualCommitConsumerDemo {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "172.16.0.218:9092,172.16.0.219:9092,172.16.0.217:9092");
        props.put("group.id", "leixiang");
        props.put("enable.auto.commit", "false");//手動確認
        /* 自動確認offset的時間間隔 */
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "earliest");//想要讀取之前的資料，必須加上
        /*
         * 一旦consumer和kakfa叢集建立連線，
         * consumer會以心跳的方式來高速叢集自己還活著，
         * 如果session.timeout.ms 內心跳未到達伺服器，伺服器認為心跳丟失，會做rebalence
         */
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //配置自定義的攔截器，可以在攔截器中引入第三方外掛實現日誌記錄等功能。
        props.put("interceptor.classes", "com.lt.kafka.consumer.MyConsumerInterceptor");

        KafkaConsumer<string, string=""> consumer = new KafkaConsumer<string, string="">(props);
        /* 消費者訂閱的topic, 可同時訂閱多個 ，用逗號隔開*/
        consumer.subscribe(Arrays.asList("leixiang"));
        while (true) {
            ConsumerRecords<string, string=""> records = consumer.poll(100);
            for (ConsumerRecord<string, string=""> record : records) {
                //處理訊息
                saveMessage(record);
                //手動提交，並且設定Offset提交回調方法
                //consumer.commitAsync(new MyOffsetCommitCallback());
                consumer.commitAsync();
            }
        }
    }

    public static void saveMessage(ConsumerRecord<string, string=""> record){
        System.out.printf("處理訊息：offset = %d, key = %s, value = %s%n", record.offset(), record.key(),
                record.value());
    }
}

自定義Consumer攔截器

import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
/**
 * @author Thomas
 * @Description:
 * @date 22:29 2019-7-5
 */
public class MyConsumerInterceptor implements ConsumerInterceptor<string, string="">{
public void configure(Map<string,> configs) {
        System.out.println("MyConsumerInterceptor configs>>>"+configs.toString());
        }

public ConsumerRecords<string, string=""> onConsume(ConsumerRecords<string, string=""> records) {
        System.out.println("onConsume");
        return records;
        }

public void onCommit(Map<topicpartition, offsetandmetadata=""> offsets) {
        System.out.println("onCommit");
        }

public void close() {
        System.out.println("MyConsumerInterceptor is closed!");
        }
}

自定義Offset提交回調方法

import java.util.Map;

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
/**
 * @author Thomas
 * @Description:
 * @date 22:31 2019-7-5
 */
public class MyOffsetCommitCallback implements OffsetCommitCallback {

    public void onComplete(Map<topicpartition, offsetandmetadata=""> offsets, Exception exception) {
        if (offsets != null)
            System.out.println("offsets>>>" + offsets.toString());
        if (exception != null)
            exception.printStac