Kafka的常用API介紹[大數(shù)據(jù)培訓(xùn)]

更新時(shí)間:2020年01月03日13時(shí)44分來(lái)源:傳智播客瀏覽次數(shù):

傳智播客

一、消息發(fā)送

1.異步發(fā)送

1）導(dǎo)入依賴

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

2）編寫代碼
需要用到的類：
KafkaProducer：需要?jiǎng)?chuàng)建一個(gè)生產(chǎn)者對(duì)象，用來(lái)發(fā)送數(shù)據(jù)
ProducerConfig：獲取所需的一系列配置參數(shù)
ProducerRecord：每條數(shù)據(jù)都要封裝成一個(gè)ProducerRecord對(duì)象

1.不帶回調(diào)函數(shù)的API

package com.heima.kafka;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CustomProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop102:9092");//kafka集群，broker-list
        props.put("acks", "all");
        props.put("retries", 1);//重試次數(shù)
        props.put("batch.size", 16384);//批次大小
        props.put("linger.ms", 1);//等待時(shí)間
        props.put("buffer.memory", 33554432);//RecordAccumulator緩沖區(qū)大小
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), Integer.toString(i)));
        }
        producer.close();
    }
}

2.帶回調(diào)函數(shù)的API

回調(diào)函數(shù)會(huì)在producer收到ack時(shí)調(diào)用，為異步調(diào)用，該方法有兩個(gè)參數(shù)，分別是RecordMetadata和Exception，如果Exception為null，說(shuō)明消息發(fā)送成功，如果Exception不為null，說(shuō)明消息發(fā)送失敗。
注意：消息發(fā)送失敗會(huì)自動(dòng)重試，不需要我們?cè)诨卣{(diào)函數(shù)中手動(dòng)重試。

package com.heima.kafka;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CustomProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop102:9092");//kafka集群，broker-list
        props.put("acks", "all");
        props.put("retries", 1);//重試次數(shù)
        props.put("batch.size", 16384);//批次大小
        props.put("linger.ms", 1);//等待時(shí)間
        props.put("buffer.memory", 33554432);//RecordAccumulator緩沖區(qū)大小
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), Integer.toString(i)), new Callback() {

                //回調(diào)函數(shù)，該方法會(huì)在Producer收到ack時(shí)調(diào)用，為異步調(diào)用
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        System.out.println("success->" + metadata.offset());
                    } else {
                        exception.printStackTrace();
                    }
                }
            });
        }
        producer.close();
    }
}

2.同步發(fā)送

同步發(fā)送的意思就是，一條消息發(fā)送之后，會(huì)阻塞當(dāng)前線程，直至返回ack。
由于send方法返回的是一個(gè)Future對(duì)象，根據(jù)Futrue對(duì)象的特點(diǎn)，我們也可以實(shí)現(xiàn)同步發(fā)送的效果，只需在調(diào)用Future對(duì)象的get方發(fā)即可。

package com.heima.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CustomProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop102:9092");//kafka集群，broker-list
        props.put("acks", "all");
        props.put("retries", 1);//重試次數(shù)
        props.put("batch.size", 16384);//批次大小
        props.put("linger.ms", 1);//等待時(shí)間
        props.put("buffer.memory", 33554432);//RecordAccumulator緩沖區(qū)大小
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), Integer.toString(i))).get();
        }
        producer.close();
    }
}

二、消息消費(fèi)

Consumer消費(fèi)數(shù)據(jù)時(shí)的可靠性是很容易保證的，因?yàn)閿?shù)據(jù)在Kafka中是持久化的，故不用擔(dān)心數(shù)據(jù)丟失問(wèn)題。
由于consumer在消費(fèi)過(guò)程中可能會(huì)出現(xiàn)斷電宕機(jī)等故障，consumer恢復(fù)后，需要從故障前的位置的繼續(xù)消費(fèi)，所以consumer需要實(shí)時(shí)記錄自己消費(fèi)到了哪個(gè)offset，以便故障恢復(fù)后繼續(xù)消費(fèi)。所以offset的維護(hù)是Consumer消費(fèi)數(shù)據(jù)是必須考慮的問(wèn)題。

1.自動(dòng)提交offset

2.手動(dòng)提交offset

雖然自動(dòng)提交offset十分簡(jiǎn)介便利，但由于其是基于時(shí)間提交的，開發(fā)人員難以把握offset提交的時(shí)機(jī)。因此Kafka還提供了手動(dòng)提交offset的API。
手動(dòng)提交offset的方法有兩種：分別是commitSync（同步提交）和commitAsync（異步提交）。兩者的相同點(diǎn)是，都會(huì)將本次poll的一批數(shù)據(jù)最高的偏移量提交；不同點(diǎn)是，commitSync阻塞當(dāng)前線程，一直到提交成功，并且會(huì)自動(dòng)失敗重試（由不可控因素導(dǎo)致，也會(huì)出現(xiàn)提交失敗）；而commitAsync則沒有失敗重試機(jī)制，故有可能提交失敗。

1）同步提交offset
由于同步提交offset有失敗重試機(jī)制，故更加可靠，以下為同步提交offset的示例。

package com.heima.kafka.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;

/**
* @author liubo
*/
public class CustomComsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop102:9092");//Kafka集群
        props.put("group.id", "test");//消費(fèi)者組，只要group.id相同，就屬于同一個(gè)消費(fèi)者組
        props.put("enable.auto.commit", "false");//關(guān)閉自動(dòng)提交offset
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("first"));//消費(fèi)者訂閱主題

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);//消費(fèi)者拉取數(shù)據(jù)
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
           consumer.commitSync();//同步提交，當(dāng)前線程會(huì)阻塞知道offset提交成功
        }
    }
}

2）異步提交offset
雖然同步提交offset更可靠一些，但是由于其會(huì)阻塞當(dāng)前線程，直到提交成功。因此吞吐量會(huì)收到很大的影響。因此更多的情況下，會(huì)選用異步提交offset的方式。
以下為異步提交offset的示例：

package com.heima.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

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

/**
* @author liubo
*/
public class CustomConsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop102:9092");//Kafka集群
        props.put("group.id", "test");//消費(fèi)者組，只要group.id相同，就屬于同一個(gè)消費(fèi)者組
        props.put("enable.auto.commit", "false");//關(guān)閉自動(dòng)提交offset
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("first"));//消費(fèi)者訂閱主題

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);//消費(fèi)者拉取數(shù)據(jù)
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            consumer.commitAsync(new OffsetCommitCallback() {
                @Override
                public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                    if (exception != null) {
                        System.err.println("Commit failed for" + offsets);
                    }
                }
            });//異步提交
        }
    }
}

3.自定義存儲(chǔ)offset

Kafka 0.9版本之前，offset存儲(chǔ)在zookeeper，0.9版本之后，默認(rèn)將offset存儲(chǔ)在Kafka的一個(gè)內(nèi)置的topic中。除此之外，Kafka還可以選擇自定義存儲(chǔ)offset。
Offset的維護(hù)是相當(dāng)繁瑣的，因?yàn)樾枰紤]到消費(fèi)者的Rebalace。

當(dāng)有新的消費(fèi)者加入消費(fèi)者組、已有的消費(fèi)者推出消費(fèi)者組或者所訂閱的主題的分區(qū)發(fā)生變化，就會(huì)觸發(fā)到分區(qū)的重新分配，重新分配的過(guò)程叫做Rebalance。
消費(fèi)者發(fā)生Rebalance之后，每個(gè)消費(fèi)者消費(fèi)的分區(qū)就會(huì)發(fā)生變化。因此消費(fèi)者要首先獲取到自己被重新分配到的分區(qū)，并且定位到每個(gè)分區(qū)最近提交的offset位置繼續(xù)消費(fèi)。
要實(shí)現(xiàn)自定義存儲(chǔ)offset，需要借助ConsumerRebalanceListener，以下為示例代碼，其中提交和獲取offset的方法，需要根據(jù)所選的offset存儲(chǔ)系統(tǒng)自行實(shí)現(xiàn)。

package com.heima.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

/**
* @author liubo
*/
public class CustomConsumer {

    private static Map<TopicPartition, Long> currentOffset = new HashMap<>();

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop102:9092");//Kafka集群
        props.put("group.id", "test");//消費(fèi)者組，只要group.id相同，就屬于同一個(gè)消費(fèi)者組
        props.put("enable.auto.commit", "false");//關(guān)閉自動(dòng)提交offset
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("first"), new ConsumerRebalanceListener() {

            //該方法會(huì)在Rebalance之前調(diào)用
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                commitOffset(currentOffset);
            }

            //該方法會(huì)在Rebalance之后調(diào)用
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                currentOffset.clear();
                for (TopicPartition partition : partitions) {
                    consumer.seek(partition, getOffset(partition));//定位到最近提交的offset位置繼續(xù)消費(fèi)
                }
            }
        });

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);//消費(fèi)者拉取數(shù)據(jù)
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                currentOffset.put(new TopicPartition(record.topic(), record.partition()), record.offset());
            }
            commitOffset(currentOffset);
        }
    }

    //獲取某分區(qū)的最新offset
    private static long getOffset(TopicPartition partition) {
        return 0;
    }

    //提交該消費(fèi)者所有分區(qū)的offset
    private static void commitOffset(Map<TopicPartition, Long> currentOffset) {

    }
}

上一篇：Mysql幫助信息命令是什么意思？[大數(shù)據(jù)培訓(xùn)] 下一篇：如何通過(guò)ECharts js實(shí)現(xiàn)數(shù)據(jù)可視化面板[大數(shù)據(jù)培訓(xùn)視頻教程]