1. 引入依赖

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>org.example</groupId>
  8.     <artifactId>kafka-producer-api</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  11.     <properties>
  12.         <maven.compiler.source>8</maven.compiler.source>
  13.         <maven.compiler.target>8</maven.compiler.target>
  14.     </properties>
  16.     <dependencies>
  17.         <dependency>
  18.             <groupId>org.apache.kafka</groupId>
  19.             <artifactId>kafka-clients</artifactId>
  20.             <version>0.11.0.0</version>
  21.         </dependency>
  23.         <dependency>
  24.             <groupId>org.apache.kafka</groupId>
  25.             <artifactId>kafka_2.12</artifactId>
  26.             <version>0.11.0.0</version>
  27.         </dependency>
  28.     </dependencies>
  30. </project>

2. 消费数据

1）高级 API 消费数据

  1. package com.example.kafka.producer;
  3. import org.apache.kafka.clients.consumer.ConsumerConfig;
  4. import org.apache.kafka.clients.consumer.ConsumerRecord;
  5. import org.apache.kafka.clients.consumer.ConsumerRecords;
  6. import org.apache.kafka.clients.consumer.KafkaConsumer;
  7. import org.apache.kafka.common.serialization.StringDeserializer;
  8. import java.util.Arrays;
  9. import java.util.Iterator;
  10. import java.util.Properties;
  12. public class SeniorConsumer {
  14.     public static void main(String[] args) {
  16.         // 创建配置对象
  17.         Properties properties = new Properties();
  18.         // 集群信息
  19.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "Hadoop102:9092, Hadoop103:9092, Hadoop104:9092");
  20.         // 键序列化
  21.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  22.         // 值序列化
  23.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  24.         // 消费者组
  25.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "terry");
  26.         // 自动确认 offset
  27.         properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
  28.         // 自动确认 offset 时间间隔
  29.         properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
  31.         // 创建消费者
  32.         KafkaConsumer<String ,String> consumer = new KafkaConsumer<String, String>(properties);
  34.         // 订阅主题
  35.         consumer.subscribe(Arrays.asList("first"));
  37.         while (true) {
  38.             // 拉取数据
  39.             ConsumerRecords<String, String> records = consumer.poll(1);
  40.             Iterator<ConsumerRecord<String, String>> iterator = records.iterator();
  41.             for (ConsumerRecord record : records) {
  42.                 // 打印数据
  43.                 System.out.println(record);
  44.             }
  45.         }
  47.     }
  49. }

2）低级 API 消费数据

  1. https://blog.csdn.net/brooks1024/article/details/104906077?utm_medium=distribute.pc_aggpage_search_result.none-task-blog-2~aggregatepage~first_rank_v2~rank_aggregation-1-104906077.pc_agg_rank_aggregation&utm_term=kafka+%E4%BD%8E%E7%BA%A7%E6%B6%88%E8%B4%B9%E8%80%85&spm=1000.2123.3001.4430

3. 知识点排查

https://www.cnblogs.com/bainianminguo/p/12247158.html