Apache Kafka學習（二）-Java使用Kafka

時間 2019-11-09

原文原文鏈接

一、maven依賴java

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

二、Producerapache

2.一、producer發送消息bootstrap

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.二、自定義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、自定義消息路由規則session

自定義路由規則，能夠根據本身的須要定義消息發送到哪一個分區。自定義路由規則須要實現Partitioner。maven

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;
    }
}

三、Consumer

3.一、自動提交函數

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、手動提交this

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攔截器spa

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.printStackTrace();
    }
}