Kafka的入门级API应用

文章目录

1.Producer API
- 1.1 消息发送流程
- 1.2 异步发送API
- - 1.2.1 导入依赖
  - 1.2.2 添加log4j配置文件
  - 1.2.3 编写代码
  - - 1.2.3.1 不带回调函数的API
    - 1.2.3.2 带回调函数的API
- 1.3 自定义分区器
2.Consumer API
- 2.1 自动提交offset
- - 2.1.1 编写代码
- 2.2 手动提交offset
- - 2.2.1 同步提交offset
  - 2.2.2 异步提交offset
- 2.3 数据漏消费和重复消费分析
3.自定义Interceptor
- 3.1 拦截器原理
- 3.2 拦截器案例
- - 3.2.1 需求
  - 3.2.2 增加时间戳拦截器
  - 3.2.3 计数拦截器
  - 3.2.4 Producer主程序
  - 3.2.5 测试

1.Producer API 1.1 消息发送流程

Kafka的Producer发送消息采用异步发送的方式。在消息发送的过程中，涉及到了两个线程—-main线程和Sender线程，以及一个线程共享变量—-RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。

下图为KafkaProducer发送消息流程：

注意：

两个重要参数：
1.batch.size:sender发送数据的前提是数据积累到batch.size（一批数据的大小）。
2.lingertime:如果数据迟迟未达到batch.size,sender等待lingertime之后就会发送数据。

1.2 异步发送API 1.2.1 导入依赖


        
            org.apache.kafka
            kafka-clients
            2.4.1
        
        
            org.apache.logging.log4j
            log4j-slf4j-impl
            2.12.0

1.2.2 添加log4j配置文件

1.2.3 编写代码

需要用到的类：

KafkaProducer：创建一个生产者对象，用来发送数据

ProducerConfig：获取所需的一系列配置参数

ProducerRecord：每条数据都要封装成一个ProducerRecord对象

1.2.3.1 不带回调函数的API

CustomProducer.java

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;


public class CustomProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        //kafka集群：broker-list
        props.put("bootstrap.servers","yxp:9092");
        //重试次数
        props.put("retries",1);
        //批次大小
        props.put("batch.size",16384);
        //等待时间
        props.put("linger.ms",1);

        //RecordAccumilator缓冲区大小
        props.put("buffer.memory",33554432);

        //键和值的序列化器
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        KafkaProducer producer = new KafkaProducer(props);

        //发送的消息内容为：[0,20)
        for (int i=0;i<20;i++){
            producer.send(new ProducerRecord("topic_jx",String.valueOf(i),String.valueOf(i)));

        }

        producer.close();
    }
}

测试:

1.在shell中先开启consumer端：

如果没有该Topic，会警告然后自动创建，不用担心。

命令：

kafka-console-consumer.sh --bootstrap-server yxp:9092 --topic topic_jx

2.运行ConusmerProducer代码

3.查看Consumer收到的消息

1.2.3.2 带回调函数的API

回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是Recordmetadata和Exception，如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。

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

对原ConsumerProducer.java做一下修改，修改后的程序代码为：

package com.yxp.kafkareview;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.Recordmetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;


public class CustomProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        //kafka集群：broker-list
        props.put("bootstrap.servers","yxp:9092");
        //重试次数
        props.put("retries",1);
        //批次大小
        props.put("batch.size",16384);
        //等待时间
        props.put("linger.ms",1);

        //RecordAccumilator缓冲区大小
        props.put("buffer.memory",33554432);

        //键和值的序列化器
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        KafkaProducer producer = new KafkaProducer(props);

        //在每次发送的数据的send方法中调用回调函数
        for (int i=0;i<20;i++){
            producer.send(new ProducerRecord("topic_jx", String.valueOf(i), String.valueOf(i)), new Callback() {
                @Override
                public void onCompletion(Recordmetadata metadata, Exception exception) {
                    if (exception==null){
                        System.out.println("success->"+metadata.offset());
                    }else{
                        exception.printStackTrace();
                    }
                }
            });
        }
        producer.close();
    }
}

发送消息后会调用该方法打印成功与否，控制台消息如下：

这success后跟的是生产者的偏移量。由于已经发送过20条（0-19），因此从20开始

Consumer端收到的消息：

1.3 自定义分区器

MyPartitioner.java

package com.yxp.kafkareview;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;


public class MyPartitionerx implements Partitioner {

    
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        int partition=1;
        if (Integer.parseInt(value.toString())%2==0){
            System.out.println(Integer.parseInt(value.toString())%2);
            partition=0;
        }else {
            System.out.println(Integer.parseInt(value.toString())%2);
        }
        return partition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map configs) {

    }
}

在Producer对应的类中指定该程序全类名。

ConsumerProducer.java

package com.yxp.kafkareview;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;


public class CustomProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        //kafka集群：broker-list
        props.put("bootstrap.servers","yxp:9092");
        //重试次数
        props.put("retries",1);
        //批次大小
        props.put("batch.size",16384);
        //等待时间
        props.put("linger.ms",1);

        //自定义分区器
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitionerx.class.getName());

        //RecordAccumilator缓冲区大小
        props.put("buffer.memory",33554432);

        //键和值的序列化器
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        KafkaProducer producer = new KafkaProducer(props);

        for (int i=0;i<40;i++){
            producer.send(new ProducerRecord("topic_mem", String.valueOf(i)), new Callback() {
                @Override
                public void onCompletion(Recordmetadata metadata, Exception exception) {
                    if (exception==null){
                        System.out.println("success->"+metadata.offset());
                    }else{
                        exception.printStackTrace();
                    }
                }
            });
        }

        producer.close();
    }
}

2.Consumer API

由于数据在Kafka中是持久化的，所以不用担心数据丢失问题，因此，Consumer消费数据时的可靠性是可以保证的。

由于consumer在消费过程中可能会出现断电宕机等故障，consumer恢复后，需要从故障前的位置继续消费，所以consumer需要实时记录自己消费到了哪个offset，以便故障恢复后继续消费。

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

2.1 自动提交offset 2.1.1 编写代码

需要用到的类：

KafkaConsumer：创建一个消费者对象，用来消费数据。

ConsumerConfig：获取所需的一系列配置参数。

ConsumerRecord：每条数据都要封装成一个ConsumerRecord对象。

Kafka提供了自动提交offset的功能。自动提交Offset的相关参数：

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

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

CustomConsumer.java

package com.yxp.kafkareview;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;


public class CustomConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();
       //kafka集群服务器 
        props.put("bootstrap.servers", "yxp:9092");
		//消费者的组id
        props.put("group.id", "test");
       //允许自动提交 
        props.put("enable.auto.commit", "true");
       //自动提交的间隔时间 
        props.put("auto.commit.interval.ms", "1000");
		//key的反序列化器        
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
       //value的反序列化器 
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer consumer = new KafkaConsumer<>(props);
       //指定提交到的topic 
        consumer.subscribe(Arrays.asList("topic_mem"));
        
        //无限循环的poll数据
        while (true) {
    ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

2.2 手动提交offset

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

手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。两者的相同点是，都会将本次poll的一批数据最高的偏移量提交；不同点是，commitSync阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而commitAsync则没有失败重试机制，故有可能提交失败。

2.2.1 同步提交offset

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

CustomConsumer.java

package com.yxp.kafkareview;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
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", "yxp:9092");
        //组id
        props.put("group.id", "test");
        //关闭自动提交
        props.put("enable.auto.commit", "false");
        //自动提交间隔
        props.put("auto.commit.interval.ms", "1000");

        //key和value的反序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Arrays.asList("topic_mem"));

        while (true) {

            ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));

            for (ConsumerRecord record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            //消费完之后同步提交offset偏移量
            consumer.commitSync();
        }


    }
}

2.2.2 异步提交offset

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

CustomConsumer.java

package com.yxp.kafkareview;

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

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


public class CustomConsumer {

    public static void main(String[] args) {
        Properties props = new Properties();

        //集群服务器
        props.put("bootstrap.servers", "yxp:9092");
        //组id
        props.put("group.id", "test");
        //关闭自动提交
        props.put("enable.auto.commit", "false");
        //自动提交间隔
        props.put("auto.commit.interval.ms", "1000");

        //key和value的反序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Arrays.asList("topic_mem"));

        while (true) {

            ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));

            for (ConsumerRecord record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            //消费完之后异步提交offset偏移量
            consumer.commitAsync(new OffsetCommitCallback() {
                @Override
                public void onComplete(Map offsets, Exception exception) {
                    if (exception!=null){
                        exception.printStackTrace();
                    }
                }
            });
        }


    }
}

2.3 数据漏消费和重复消费分析

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

3.自定义Interceptor 3.1 拦截器原理

Producer拦截器（Interceptor）主要用于实现clients端的定制化控制逻辑。

对于Producer而言，Interceptor使得用户在消息发送前以及producer回调逻辑前有机会对消息做一些定制化需求，比如修改消息等。同时，producer允许用户指定多个Interceptor按序作用于同一条消息从而形成一个拦截链（Interceptor chain）。拦截器需要实现ProducerInterceptor接口，其定义的方法包括：

（1）configure(configs):

获取配置信息和初始化数据时调用。

（2）onSend(ProducerRecord)：

该方法封装进KafkaProducer.send方法中，即它运行在用户主线程中。Producer确保在消息被序列化以及计算分区前调用该方法。用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的topic和分区，否则会影响目标分区的计算。

（3）onAcknowledgement(Recordmetadata, Exception)：

该方法会在消息从RecordAccumulator成功发送到Kafka Broker之后，或者在发送过程中失败时调用。并且通常都是在producer回调逻辑触发之前。onAcknowledgement运行在producer的IO线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢producer的消息发送效率。

（4）close：

关闭interceptor，主要用于执行一些资源清理工作.

如前所述，interceptor可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外倘若指定了多个interceptor，则producer将按照指定顺序调用它们，并仅仅是捕获每个interceptor可能抛出的异常记录到错误日志中而非向上传递。这在使用过程中要特别留意。

3.2 拦截器案例

3.2.1 需求

实现一个简单的双interceptor组成的拦截链。第一个interceptor会在消息发送前将时间戳信息加到消息value的最前部；第二个interceptor会在消息发送后更新成功发送消息数或失败发送消息数。

3.2.2 增加时间戳拦截器

TimeInterceptor.java

package com.yxp.kafkareview;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.Recordmetadata;

import java.util.Map;


public class TimeInterceptor implements ProducerInterceptor {
    @Override
    public ProducerRecord onSend(ProducerRecord record) {

//新构建一个ProducerRecorder对象（在值属性那里将时间戳加在原值前面）
        return new ProducerRecord<>(record.topic(),record.partition(),record.key(),System.currentTimeMillis()+","+ record.value());

    }

    @Override
    public void onAcknowledgement(Recordmetadata metadata, Exception exception) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map configs) {

    }
}

3.2.3 计数拦截器

CounterInterceptor.java

package com.yxp.kafkareview;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.Recordmetadata;

import java.util.Map;


public class CounterInterceptor implements ProducerInterceptor {

    private int errorCounter=0;
    private int successCounter=0;


    @Override
    public ProducerRecord onSend(ProducerRecord record) {
        return record;
    }

    @Override
    public void onAcknowledgement(Recordmetadata metadata, Exception exception) {

        //统计成功和失败的次数
        if (exception==null){
            successCounter++;
        }else {
            errorCounter++;
        }

    }

    @Override
    public void close() {

        System.out.println("Success sent:"+successCounter);
        System.out.println("Error sent:"+errorCounter);
    }

    @Override
    public void configure(Map configs) {

    }
}

3.2.4 Producer主程序

package com.yxp.kafkareview;

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

import java.util.ArrayList;
import java.util.Properties;


public class InterceptorProducer {
    public static void main(String[] args) {

        //1.设置配置信息
        Properties props = new Properties();
        props.put("bootstrap.servers", "yxp:9092");
        props.put("acks", "all");
        props.put("retries", 3);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        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");


        

        //2.构建拦截链
        ArrayList interceptors = new ArrayList<>();
        interceptors.add(TimeInterceptor.class.getName());
        interceptors.add(CounterInterceptor.class.getName());

        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptors);


        KafkaProducer producer = new KafkaProducer<>(props);
        
        //3.发送消息
        for (int i=0;i<10;i++){
            producer.send(new ProducerRecord<>("Topic_new","message"+i));
        }

        //4.关闭producer
        producer.close();
    }

}

3.2.5 测试

1.在kafka上启动消费者

shell端命令：

kafka-console-consumer.sh --bootstrap-server yxp:9092 --topic Topic_new

2.然后运行客户端java程序。

3.观察消费者接受到的信息

Kafka的入门级API应用

大数据系统相关栏目本月热门文章