快速入门Kafka系列(6)——Kafka的JavaAPI操作

        作为快速入门Kafka系列的第六篇博客，本篇为大家带来的是Kafka的JavaAPI操作~

        码字不易,先赞后看!

文章目录

• • 1. 创建Maven工程并添加jar包
  • 2. 生产者代码
  • 4. Kafka Streams API开发
  • 3.1 自动提交offset
  • 3.2 手动提交offset
  • 3.3 消费完每个分区之后手动提交offset
  • 3.4 指定分区数据进行消费
  • 3.5 重复消费与数据丢失
  • 1. 使用生产者，生产数据
  • 2. kafka当中的数据分区
  • 3. 消费者代码
  • 4.1 创建一个Topic
  • 4.2 开发StreamsAPI
  • Kafka的JavaAPI操作

---

Kafka的JavaAPI操作

1. 创建Maven工程并添加jar包

        首先在IDEA中我们创建一个maven工程，并添加以下依赖的jar包的坐标到pom.xml

 <dependencies>                  <dependency>             <groupId>org.apache.kafkagroupId>             <artifactId>kafka-clientsartifactId>             <version>2.0.0version>         dependency>         <dependency>             <groupId>org.apache.kafkagroupId>             <artifactId>kafka-streamsartifactId>             <version>2.0.0version>         dependency>     dependencies>     <build>         <plugins>                          <plugin>                 <groupId>org.apache.maven.pluginsgroupId>                 <artifactId>maven-compiler-pluginartifactId>                 <version>3.2version>                 <configuration>                     <source>1.8source>                     <target>1.8target>                     <encoding>UTF-8encoding>                 configuration>             plugin>         plugins>     build>     <repositories>         <repository>             <id>maven-aliid>             <url>http://maven.aliyun.com/nexus/content/groups/public//url>             <releases>                 <enabled>trueenabled>             releases>             <snapshots>                 <enabled>trueenabled>                 <updatePolicy>alwaysupdatePolicy>                 <checksumPolicy>failchecksumPolicy>             snapshots>         repository>     repositories>

2. 生产者代码

1. 使用生产者，生产数据

/* 用于生产数据到kafka集群  */public class Producer1 {     /*     程序的入口      */     public static void main(String[] args){         //编写生产数据的程序         //1、配置kafka集群环境（设置）         Properties props = new Properties();         //kafka服务器地址         props.put("bootstrap.servers", "node01:9092,node02:9092,node03:9092");         //消息确认机制         props.put("acks", "all");         //重试机制         props.put("retries", 0);         //批量发送的大小         props.put("batch.size", 16384);         //消息延迟         props.put("linger.ms", 1);         //批量的缓冲区大小         props.put("buffer.memory", 33554432);         // kafka   key 和value的序列化         props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");         props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");         //2、实例一个生产者对象         KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);         for (int i = 0; i < 9; i++) {             //3、发送数据 ,需要一个producerRecord对象,最少参数 String topic, V value              //ProducerRecordrecord             ProducerRecord producerRecord = new ProducerRecord<>("18BD12","bbbb___"+i);                          //4、通过生产者对象将数据发送到kafka集群             kafkaProducer.send(producerRecord);         }         //4、关闭成产者         kafkaProducer.close();     }}

2. kafka当中的数据分区

        kafka生产者发送的消息，都是保存在broker当中，我们可以自定义分区规则，决定消息发送到哪个partition里面去进行保存。

        查看ProducerRecord这个类的源码，就可以看到kafka的各种不同分区策略。

        kafka当中支持以下四种数据的分区方式：

       //1、没有指定分区编号，没有指定key,时采用轮询方式存户数据        ProducerRecord producerRecord = new ProducerRecord<>("18BD12","bbbb___"+i);                     //2、没有指定分区编号，指定key时，数据分发策略为对key求取hash值，这个值与分区数量取余，于数就是分区编号。        //ProducerRecord producerRecord = new ProducerRecord("18BD12","test","aaaa___"+i);                    //3、指定分区编号，所有数据输入到指定的分区内        //ProducerRecord producerRecord = new ProducerRecord("18BD12",1,"test","aaaa___"+i);        //4、自定义分区策略。如果不自定义分区规则，那么会将数据使用轮询的方式均匀的发送到各个分区里面去        //ProducerRecord producerRecord = new ProducerRecord("18BD12","test","aaaa___"+i);

        其中，自定义分区策略需要我们单独创建一个类，并在类中定义我们所想要的分区规则。

public class KafkaCustomPartitioner implements Partitioner { @Override public void configure(Map<String, ?> configs) { } @Override public int partition(String topic, Object arg1, byte[] keyBytes, Object arg3, byte[] arg4, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);     int partitionNum = partitions.size(); Random random = new Random(); int partition = random.nextInt(partitionNum);     return partition; } @Override public void close() { }}

        并在主代码中添加配置,其中partitioner.class的值对应的就是我们单独写的一个实现Partitioner 的类在项目中具体带包名的路径

props.put("partitioner.class", "com.czxy.demo_test.Demo05.KafkaCustomPartitioner ");

        我们也可以通过IDEA中的快捷键来实现快速获取

3. 消费者代码

        消费者要从kafka Cluster进行消费数据，必要条件有以下四个

#1、地址 bootstrap.servers=node01:9092#2、序列化  key.serializer=org.apache.kafka.common.serialization.StringSerializer value.serializer=org.apache.kafka.common.serialization.StringSerializer #3、主题（topic） 需要制定具体的某个topic（order）即可。 #4、消费者组 group.id=test

3.1 自动提交offset

消费完成之后，自动提交offset

public class Consumer01 {     public static void main(String[] args) {         //1、添加配置文件         Properties props = new Properties();         //指定kafka服务器         props.put("bootstrap.servers", "node01:9092,node02:9092,node03:9092");         //消费组         props.put("group.id", "test");         //以下两行代码 ---消费者自动提交offset值         props.put("enable.auto.commit", "true");         //自动提交的周期         props.put("auto.commit.interval.ms",  "1000");         //kafka   key 和value的反序列化         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");         //2、实例消费者对象         KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);         //3、设置读取的topic         kafkaConsumer.subscribe(Arrays.asList("student"));         //循环遍历         while (true){             //4、拉取数据，并输出             //获取到所有的数据             ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(1000);             //遍历所有数据，获得到一条             for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {                 //一条数据                 System.out.println("当前数据："+consumerRecord.value()+", 偏移量:offset:"+consumerRecord.offset());             }         }     }}

3.2 手动提交offset

        如果Consumer在获取数据后，需要加入处理，数据完毕后才确认offset，需要程序来控制offset的确认，该如何操作？

        我们就需要在配置kafka环境配置的时候关闭自动提交确认选项

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

        然后在循环遍历消费的过程中，消费完毕就手动提交。

  while (true){             //4、拉取数据，并输出             ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(1000);             //遍历所有数据获取一条             for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {                 System.out.println(consumerRecord.value()  +"     "+consumerRecord.offset());             }             //手动提交offset             kafkaConsumer.commitSync();         }

        或者也可以将手动提交offset的语句放置到循环体中，每消费一条数据，就手动提交一次offset也是可以的。

3.3 消费完每个分区之后手动提交offset

        上面的示例使用commitSync将所有已接收的记录标记为已提交。 在某些情况下，您可能希望通过明确指定偏移量 来更好地控制已提交的记录。 在下面的示例中，我们在完成处理每个分区中的记录后提交偏移量。

try {while(running) {ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE); for (TopicPartition partition : records.partitions()) {List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);for (ConsumerRecord<String, String> record : partitionRecords) { System.out.println(record.offset() + ": " + record.value());}long lastOffset = partitionRecords.get(partitionRecords.size() -1).offset();consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));}}} finally { consumer.close();}

注意事项：

        提交的偏移量应始终是应用程序将读取的下一条消息的偏移量。 因此，在调用commitSync（偏移量）时，应该 在最后处理的消息的偏移量中添加一个。

3.4 指定分区数据进行消费

        1、如果进程正在维护与该分区关联的某种本地状态（如本地磁盘上的键值存储），那么它应该只获取它在磁盘上 维护的分区的记录。

        2、如果进程本身具有高可用性，并且如果失败则将重新启动（可能使用YARN，Mesos或AWS工具等集群管理框 架，或作为流处理框架的一部分）。 在这种情况下，Kafka不需要检测故障并重新分配分区，因为消耗过程将在另 一台机器上重新启动。

       // 第一个参数为消费的Topic,第二个参数为消费的Partition         TopicPartition topicPartition0 = new TopicPartition("18BD12",0);         TopicPartition topicPartition2 = new TopicPartition("18BD12",1);         kafkaConsumer.assign(Arrays.asList(topicPartition0,topicPartition2));

注意事项：

        1、要使用此模式，您只需使用要使用的分区的完整列表调用assign（Collection），而不是使用subscribe订阅主题。

        2、主题与分区订阅只能二选一

3.5 重复消费与数据丢失

说明：

• 已经消费的数据对于kafka来说，会将消费组里面的offset值进行修改，那什么时候进行修改了？是在数据消费 完成之后，比如在控制台打印完后自动提交；

• 提交过程：是通过kafka将offset进行移动到下个message所处的offset的位置。

• 拿到数据后，存储到hbase中或者mysql中，如果hbase或者mysql在这个时候连接不上，就会抛出异常，如果在处理数据的时候已经进行了提交，那么kafka上的offset值已经进行了修改了，但是hbase或者mysql中没有数据，这个时候就会出现数据丢失。

• 什么时候提交offset值？在Consumer将数据处理完成之后，再来进行offset的修改提交。默认情况下offset是 自动提交，需要修改为手动提交offset值。

• 如果在处理代码中正常处理了，但是在提交offset请求的时候，没有连接到kafka或者出现了故障，那么该次修 改offset的请求是失败的，那么下次在进行读取同一个分区中的数据时，会从已经处理掉的offset值再进行处理一 次，那么在hbase中或者mysql中就会产生两条一样的数据，也就是数据重复

4. Kafka Streams API开发

        需求：使用StreamAPI获取test这个topic当中的数据，然后将数据全部转为大写，写入到test2这个topic当中去。

4.1 创建一个Topic

node01服务器使用以下命令来常见一个topic 名称为test2

cd /export/servers/kafka_2.11-1.0.0/bin/kafka-topics.sh --create  --partitions 3 --replication-factor 2 --topic test2 --zookeeper node01:2181,node02:2181,node03:2181

4.2 开发StreamsAPI

public class KafkaStream {     public static void main(String[] args) {         Properties props = new Properties();         //设置程序的唯一标识         props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application12");         //设置kafka集群         props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092,node02:9092,Node03:9092");         //设置序列化与反序列化         props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());         props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());         //实例一个计算逻辑         StreamsBuilder streamsBuilder = new StreamsBuilder();         //设置计算逻辑   stream 在哪里读取数据                ->                               to 将数据写入哪里         streamsBuilder.stream("test").mapValues(line->line.toString().toUpperCase()).to("test2");         //构建Topology对象（拓扑，流程）         final Topology topology = streamsBuilder.build();         //实例 kafka流         KafkaStreams streams = new KafkaStreams(topology, props);         //启动流计算         streams.start();     }}

        上述代码成功运行之后，就可以做到实时将test中生产的数据写入到test2中。

---

        
        好了，本次的知识分享就到这里，受益或对大数据技术感兴趣的朋友记得点赞关注一下哟~，下一篇博客将为大家介绍的是Kafka的log存储机制与消息不丢失机制，敬请期待|ू･ω･` )