Producer API

Kafka的Producer发送消息采用的是异步发送的方式。
在消息发送的过程中，涉及到了两个线程——**main线程和Sender线程，main线程将消息发送给RecordAccumulator
以及一个线程共享变量——**RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。
image.png
相关参数说明：
batch.size：只有数据积累到batch.size之后，sender才会发送数据。
linger.ms：如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据

异步发送API

1）导入依赖

  1. <dependency>
  2. <groupId>org.apache.kafka</groupId>
  3. <artifactId>kafka-clients</artifactId>
  4. <version>2.4.1</version>
  5. </dependency>

2）编写代码
需要用到的类：
KafkaProducer：需要创建一个生产者对象，用来发送数据
ProducerConfig：获取所需的一系列配置参数
ProducerRecord：每条数据都要封装成一个ProducerRecord对象

image.png

（1）不带回调函数的API

*注意：以下对比官网代码增加了几个参数

  1. package com.atguigu.kafka;
  3. import org.apache.kafka.clients.producer.*;
  5. import java.util.Properties;
  6. import java.util.concurrent.ExecutionException;
  8. public class CustomProducer {
  10.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  11.         Properties props = new Properties();
  12.        props.put("bootstrap.servers", "hadoop102:9092");//kafka集群，broker-list
  13.        props.put("acks", "all");
  14.        props.put("retries", 1);//重试次数（一般重复次数默认是3次）
  15.        props.put("batch.size", 16384);//批次大小，只要不带单位都是字节，所以是16k，每16k拉一次
  16.        props.put("linger.ms", 1);//等待时间
  17.        props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小（这个是默认值）
  18.        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  19.        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  21.         Producer<String, String> producer = new KafkaProducer<>(props);
  22.         for (int i = 0; i < 100; i++) {
  23.             producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), Integer.toString(i)));//主题+key+value
  24.         }
  25.         producer.close();
  26.     }
  27. }

（2）带回调函数的API

回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是RecordMetadata和Exception，如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。
注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

主要作用：
因为不希望在其他地方调用显示信息，会涉及到传递的问题，会导致效率低、容易报错。 —>最好找其他老师的视频看看
所以这个函数可以：在API内部判断是否成功发送、并打印信息

//

package com.atguigu.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);//重试次数
       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<>(props);
        for (int i = 0; i < 100; 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();
    }
}

同步发送API

同步发送的意思就是，一条消息发送之后，会阻塞当前线程，直至返回ack。
由于send方法返回的是一个Future对象，根据Futrue对象的特点，我们也可以实现同步发送的效果，只需在调用Future对象的get方发即可。

与异步相比：只多了个get（）方法： producer.send(new ProducerRecord(“first”, Integer.toString(i), Integer.toString(i))).get();
同步相对安全些，效率慢些
/

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

Comsumer API

Consumer消费数据时的可靠性是很容易保证的，因为数据在Kafka中是持久化的，故不用担心数据丢失问题。
由于consumer在消费过程中可能会出现断电宕机等故障，consumer恢复后，需要从故障前的位置的继续消费，所以consumer需要实时记录自己消费到了哪个offset，以便故障恢复后继续消费。
所以offset的维护是Consumer消费数据是必须考虑的问题。

自动提交offset

1）导入依赖

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


/
2）编写代码
需要用到的类：
KafkaConsumer：需要创建一个消费者对象，用来消费数据
ConsumerConfig：获取所需的一系列配置参数
ConsuemrRecord：每条数据都要封装成一个ConsumerRecord对象
为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。
自动提交offset的相关参数：
enable.auto.commit：是否开启自动提交offset功能
auto.commit.interval.ms：自动提交offset的时间间隔
以下为自动提交offset的代码：

/

package com.atguigu.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");//一般300
       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);//官方文档是Duration.ofMillis(100)每100毫秒poll（获取）一次
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}

手动提交offset

虽然自动提交offset十分简介便利，但由于其是基于时间提交的，开发人员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API。
手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。
两者的相同点是，都会将本次**poll的一批数据最高的偏移量提交**；
不同点是：
commitSync阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；
commitAsync则没有失败重试机制，故有可能提交失败。

1）同步提交offset
由于同步提交offset有失败重试机制，故更加可靠，以下为同步提交offset的示例。

/

package com.atguigu.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");//消费者组，只要group.id相同，就属于同一个消费者组
       props.put("enable.auto.commit", "false");//关闭自动提交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"));//消费者订阅主题

        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());
            }
            consumer.commitSync();//同步提交，当前线程会阻塞知道offset提交成功，相比自动提交只多了这个
        }
    }
}

2）异步提交offset
虽然同步提交offset更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会收到很大的影响。因此更多的情况下，会选用异步提交offset的方式。
以下为异步提交offset的示例：

//

package com.atguigu.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");//消费者组，只要group.id相同，就属于同一个消费者组
       props.put("enable.auto.commit", "false");//关闭自动提交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"));//消费者订阅主题

        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());
            }
           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，有可能会造成数据的重复消费。
image.png