kafka java API的使用

　　Kafka包含四种核心的API:

　　1、Producer API支持应用将数据流发送到Kafka集群的主题

　　2、Consumer API支持应用从Kafka集群的主题中读取数据流

　　3、Streams API支持数据流从输入主题转化到输出主题

　　4、Connect API支持实现持续地从一些源系统或应用划入Kafka或者从Kafka推入一些源系统或应用的接口。

　　我们这里主要讨论Producer API和Consumer API的使用，由于最新版的kafka java api中使用了一些jdk8的新特性，所以要求我们在本机上jdk版本要在8以上。

　　pom.xml如下：

<dependency>

            <groupId>org.springframework.kafka</groupId>

            <artifactId>spring-kafka</artifactId>

        </dependency>

Producer API

　　Producer用来向Kafka集群中发布消息记录的Kafka客户端。Producer是线程安全的，并且通常来讲，在多个线程间共享一个producer要比每个线程都创建一个producer速度更快。producer代码示例：

package com.example.demo;

import java.util.Properties;

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

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

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

public class MyProducer {

    public static void main(String[] args) {

        Properties props = new Properties();

        props.put("bootstrap.servers", "192.168.1.124:9092");

                props.put("acks", "all");

                props.put("retries", 0);

                props.put("batch.size", 16384);

                props.put("linger.ms", 1);

                props.put("partitioner.class", "com.example.demo.MyPartitioner");

                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<>(props);

        for (int i = 0; i < 100; i++)

            producer.send(new ProducerRecord<String, String>("powerTopic", Integer.toString(i), Integer.toString(i)));

        producer.close();

    }

}

　　properties里用到的配置参数在kafka的源码里org.apache.kafka.clients.producer.ProducerConfig类中，这里说一下常用的：

　　bootstrap.servers 配置项处需要填写我们要发送到的Kafka集群地址。

　　ack 配置项用来控制producer要求leader确认多少消息后返回调用成功。当值为0时producer不需要等待任何确认消息。当值为1时只需要等待leader确认。当值为-1或all时需要全部ISR集合返回确认才可以返回成功。

　　retries 当 retries > 0 时，如果发送失败，会自动尝试重新发送数据。发送次数为retries设置的值。

　　buffer.memory、batch.size、linger.ms三个参数用来控制缓冲区大小和延迟发送时间，具体含义可以参考官方文档的配置。

　　key.serializer 和 value.serializer 指定使用什么序列化方式将用户提供的key和value进行序列化。

Consumer API

　　Consumer的API分为High-level API和Low-level API。前者提供了高度抽象的API，使用起来简单、方便。因此本文将主要讲述High-level API。Low-level API提供了更强的控制能力，但使用起来较为繁琐。自动提交consumer代码示例：

package com.example.demo;

import java.util.Arrays;

import java.util.Properties;

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

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

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

public class MyAutoCommitConsumer {

    public static void main(String[] args) {

         Properties props = new Properties();

         props.put("bootstrap.servers", "192.168.1.124: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");

         @SuppressWarnings("resource")

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

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

         while (true) {

             ConsumerRecords<String, String> records = consumer.poll(100);

             for (ConsumerRecord<String, String> record : records)

                 System.out.printf("partition = %d,offset = %d, key = %s, value = %s%n",record.partition(), record.offset(), record.key(), record.value());

         }

    }

}

　　properties里用到的配置参数在kafka的源码里org.apache.kafka.clients.consumer.ConsumerConfig类中，本例中用到参数解释如下：

　　bootstrap.servers配置项指定了consumer需要连接的服务器集群。多台服务器用“,”分隔

　　enable.auto.commit配置项指定了提交offset的方式为自动提交，auto.commit.interval.ms配置项配置了每次自动提交的时间间隔。

　　group.id 即消费者组标签，本例中消费者组的名称为test。

　　key.deserializer和value.deserializer指用什么方式进行反序列化。

　　自动提交offset的方式非常简单，但多数情况下，我们不会使用自动提交的方式。因为不论从Kafka集群中拉取的数据是否被处理成功，offset都会被更新，也就是如果处理过程中出现错误可能会出现数据丢失的情况。所以多数情况下我们会选择手动提交方式，我们看到 enable.auto.commit 配置项被设置为false，代表手动提交。示例代码如下：

package com.example.demo;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.List;

import java.util.Properties;

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

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

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

public class MyManualCommitConsumer {

    public static void main(String[] args) {

         Properties props = new Properties();

         props.put("bootstrap.servers", "192.168.1.124:9092");

         props.put("group.id", "test");

         props.put("enable.auto.commit", "false");

         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

         @SuppressWarnings("resource")

         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

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

         final int minBatchSize = 200;

         List<ConsumerRecord<String, String>> list = new ArrayList<>();

         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());

                 list.add(record);

             }

             if (list.size() >= minBatchSize) {

                 System.out.println("list中的缓存数据大于minBatchSize时批量进行处理");

                 consumer.commitSync();

                 System.out.println("全部数据处理成功后手动提交");

                 list.clear();

             }

         }

    }

}

　　另外需注意，consumer是有状态的，所以不是线程安全的，所以在进行多线程操作时需要在每个线程实例化一个consumer。

kafka java API的使用的相关教程结束。