四、kafka API

1、Producer API

1）消息发送流程

Kafka 的 Producer 发送消息采用的是异步发送的方式。在消息发送的过程中，涉及到了 两个线程——main 线程和 Sender 线程，

以及一个线程共享变量——RecordAccumulator。 main 线程将消息发送给 RecordAccumulator，Sender 线程不断从 RecordAccumulator

中拉取 消息发送到 Kafka broker

2）异步发送API

①、依赖

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

②、不带回调函数的api

public class MyProducer {

    public static void main(String[] args) {
        Properties props = new Properties();
        //kafka 集群，broker-list
        props.put("bootstrap.servers", "121.40.182.123:9092");
        props.put("acks", "all");
        //重试次数
        props.put("retries", 3);
        //批次大小: 只有数据积累到 batch.size 之后，sender 才会发送数据
        props.put("batch.size", 16384);
        //等待时间: 如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据
        props.put("linger.ms", 1);
        //RecordAccumulator 缓冲区大小
        props.put("buffer.memory", 33554432);
        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<String, String>(props);
        for (int i = 0; i < 10; i++) {
            System.out.println(i);
            producer.send(new ProducerRecord<String, String>("first",
                    "atguigu"+i));
        }
        producer.close();
    }
}

用kafka的命令行方式启动消费者来消费消息

bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --from-beginning --topic first

可以看到消费到生产的消息

③ 带回调函数的api

回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是 RecordMetadata 和 Exception，如果 Exception 为 null，

说明消息发送成功，如果 Exception 不为 null，说明消息发送失败。

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试

public class MyCallBackProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "121.40.182.123:9092");//kafka 集群，broker-list
        props.put("acks", "all");
        props.put("retries", 1);//重试次数
        props.put("batch.size", 16384);//批次大小
        props.put("linger.ms", 1);//等待时间
        props.put("buffer.memory", 33554432);//RecordAccumulator 缓 冲区大小
        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<String, String>(props);

        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<String, String>("first",
                    Integer.toString(i), Integer.toString(i)), new Callback() {
                //回调函数，该方法会在 Producer 收到 ack 时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata metadata,
                                         Exception exception) {
                    if (exception == null) {
                        System.out.println("success->" +
                                metadata.offset());
                    } else {
                        exception.printStackTrace();
                    }
                }
            });
        }
        producer.close();
    }
}

idea控制台

消费者消费到的消息

④【扩展】自定义分区的生产者

自定义分区器

public class MyPartitioner implements Partitioner {
    @Override
    public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) {
        return 1;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}

在生产者中指定分区器

//添加分区器
props.put("partitioner.class", "com.atguigu.producer.MyPartitioner")

public class MyPartitionerProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        //kafka 集群，broker-list
        props.put("bootstrap.servers", "121.40.182.123:9092");
        props.put("acks", "all");
        //重试次数
        props.put("retries", 3);
        //批次大小: 只有数据积累到 batch.size 之后，sender 才会发送数据
        props.put("batch.size", 16384);
        //等待时间: 如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据
        props.put("linger.ms", 1);
        //RecordAccumulator 缓冲区大小
        props.put("buffer.memory", 33554432);
        props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        //添加分区器
        props.put("partitioner.class", "com.atguigu.producer.MyPartitioner");

        //创建生产者对象
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<String, String>("first",
                    "atguigu"+i) , new Callback() {
                //回调函数，该方法会在 Producer 收到 ack 时调用，为异步调用
                @Override
                public void onCompletion(RecordMetadata metadata,
                                         Exception exception) {
                    if (exception == null) {
                        System.out.println("success->" +
                                metadata.offset());
                    } else {
                        exception.printStackTrace();
                    }
                }
            });
        }
        producer.close();
    }
}

3）同步发送 API

在sender线程发送消息时阻塞 main线程，当发送完消息后，再放开main线程

public class MyProducer1 {
    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);//重试次数
        props.put("batch.size", 16384);//批次大小
        props.put("linger.ms", 1);//等待时间
        props.put("buffer.memory", 33554432);//RecordAccumulator 缓冲区大小

        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<String, String>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("first",
                    Integer.toString(i), Integer.toString(i))).get();
        }
        producer.close();

    }
}

2、consumer API

Consumer消费数据时的可靠性是很容易保证的，因为数据在Kafka中是持久化的，故不用担心数据丢失问题

由于consumer在消费过程中可能出现断电宕机等故障，consumer恢复后，需要从故障前的位置继续消费，所以

consumer需要实时记录自己消费到了哪个offset，以便故障恢复后继续消费

所以 offset的维护是Consumer消费数据必须考虑的问题

1）自动提交 offset

为了使我们能够专注于自己的业务逻辑，Kafka 提供了自动提交 offset 的功能。

自动提交 offset 的相关参数

enable.auto.commit：是否开启自动提交 offset 功能

auto.commit.interval.ms：自动提交 offset 的时间间隔

public class MyProducer {

    public static void main(String[] args) {
        Properties props = new Properties();
        //kafka 集群，broker-list
        props.put("bootstrap.servers", "121.40.182.123:9092");
        props.put("acks", "all");
        //重试次数
        props.put("retries", 3);
        //批次大小: 只有数据积累到 batch.size 之后，sender 才会发送数据
        props.put("batch.size", 16384);
        //等待时间: 如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据
        props.put("linger.ms", 1);
        //RecordAccumulator 缓冲区大小
        props.put("buffer.memory", 33554432);
        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<String, String>(props);
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<String, String>("first",
                    "atguigu"+i));
        }
        producer.close();
    }
}

利用上面的生产者生产消息，此消费者可以消费到数据

【扩展】

消费者重置 offset

//重置消费者的offset

properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

【注意】此属性生效是有条件的

2）手动提交offset

虽然自动提交 offset 十分简介便利，但由于其是基于时间提交的，开发人员难以把握 offset 提交的时机。因此 Kafka 还提供了手动提交 offset 的 API。

手动提交 offset 的方法有两种：分别是 commitSync（同步提交）和 commitAsync（异步 提交）。

两者的相同点是，都会将本次 poll 的一批数据最高的偏移量提交；

不同点是， commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致， 也会出现提交失败）；而 commitAsync 则没有失败重试机制，

故有可能提交失败

①、 同步提交

需要先关闭自动提交的属性

public class MyConsumer {

    public static void main(String[] args) {

        //1、创建消费者的配置信息
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "121.40.182.123:9092");
        //开启自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 1s中提交一次offset
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //k v 的反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        //设置消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "bigdata3");

        //重置消费者的offset
       // properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        //订阅主题
        consumer.subscribe(Arrays.asList("first"));
        while (true){
            //拉取数据
            ConsumerRecords<String, String> consumerRecords = consumer.poll(100);
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord.key() + " ---> " + consumerRecord.value());
            }

            //同步提交，当前线程会阻塞到offset成功
            consumer.commitSync();
        }
    }
}

②、异步提交

public class MyConsumer2 {

    public static void main(String[] args) {

        //1、创建消费者的配置信息
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "121.40.182.123:9092");
        //开启自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 1s中提交一次offset
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //k v 的反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        //设置消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "bigdata");

        //重置消费者的offset
       // properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        //订阅主题
        consumer.subscribe(Arrays.asList("first"));
        while (true){
            //拉取数据
            ConsumerRecords<String, String> consumerRecords = consumer.poll(100);
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord.key() + " ---> " + consumerRecord.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);
                    }
                }
            });

        }
    }
}

③、消费漏数据和重复消费

无论是同步提交还是异步提交 offset，都有可能会造成数据的漏消费或者重复消费。

先 提交 offset 后消费，若此时服务挂掉，则会丢失数据

先消费后提交 offset，若此时服务挂掉，重启后有可能会造成数据 的重复消费。

3）自定义存储offset

====== 还没有总结 ========