Kafka常用的API有哪些？

Kafka是什么?Kafka最初由LinkedIn開(kāi)發(fā)，是一款基于分區(qū)、多副本的分布式控制器，基于ZooKeeper協(xié)調(diào)。它最大的特點(diǎn)是能夠?qū)崟r(shí)處理大量數(shù)據(jù)，滿足各種需求場(chǎng)景：如基于hadoop的批處理系統(tǒng)、低延遲實(shí)時(shí)系統(tǒng)、storm/spark流媒體引擎、web/nginx日志、訪問(wèn)日志、消息服務(wù)等，采用scala語(yǔ)言編寫。LinkedIn在2010貢獻(xiàn)了Apache基金會(huì)，并成為頂級(jí)開(kāi)源項(xiàng)目。

Kafka常用的API介紹：

一、消息發(fā)送

1、異步發(fā)送

·導(dǎo)入依賴

<dependency>

<groupId>org.apache.kafka</groupId>

<artifactId>kafka-clients</artifactId>

<version>0.11.0.0</version>

</dependency>

·編寫代碼

需要用到的類：

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

1)導(dǎo)入依賴

<dependency>

<groupId>org.apache.kafka</groupId>

<artifactId>kafka-clients</artifactId>

<version>0.11.0.0</version>

</dependency>

2)編寫代碼

需要用到的類：

KafkaConsumer：需要?jiǎng)?chuàng)建一個(gè)消費(fèi)者對(duì)象，用來(lái)消費(fèi)數(shù)據(jù)

ConsumerConfig：獲取所需的一系列配置參數(shù)

ConsuemrRecord：每條數(shù)據(jù)都要封裝成一個(gè)ConsumerRecord對(duì)象

為了使我們能夠?qū)Ｗ⒂谧约旱臉I(yè)務(wù)邏輯，Kafka提供了自動(dòng)提交offset的功能。

自動(dòng)提交offset的相關(guān)參數(shù)：

enable.auto.commit：是否開(kāi)啟自動(dòng)提交offset功能

auto.commit.interval.ms：自動(dòng)提交offset的時(shí)間間隔

以下為自動(dòng)提交offset的代碼：

package com.heima.kafka;
 
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 CustomConsumer {
 
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop102:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        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"));
        while (true) {
            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());
        }
    }
}

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

雖然自動(dòng)提交offset十分簡(jiǎn)介便利，但由于其是基于時(shí)間提交的，開(kāi)發(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則沒(méi)有失敗重試機(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) {
 
    }
}

以上我們介紹了Kafka的常用API希望對(duì)您有所幫助，如果想了解更多大數(shù)據(jù)相關(guān)知識(shí)，請(qǐng)點(diǎn)擊頁(yè)面咨詢按鈕，獲得幫助。

猜你喜歡

什么是Docker？有什么特點(diǎn)？

Spark與Hadoop有哪些區(qū)別？

冒泡排序算法[動(dòng)圖介紹]