KAFKA-03-kafka基本API使用

川长思鸟来 2023-09-29 13:57 30阅读 0赞

### 文章目录 ###

*  1:topic操作api
 *   *  1：pom.xml导包
     *  2：创建topic
     *  3：查看topic列表
     *  4：查看topic详情
     *  5：删除topic
 *  2：生产者和消费者
 *   *  1：生产者
     *   *  1：普通异步发送
         *  2：带回调函数的异步发送
         *  3:同步发送
         *  4:prop参数名称 描述
     *  2：消费者
     *   *  1:消费者组初始化流程
         *  2：参数名称 描述
         *  3:订阅主题
         *  4：订阅分区
 *  3：自定义分区
 *   *  1：系统默认分区策略
     *  2：自定义分区策略
     *   *  1：分区策略默认实现Partitioner
         *  2:使用自定义分区策略
 *  4:序列化（重要）
 *   *  1：系统默认的序列化器
     *  2：自定义序列化
     *   *  1：自定义序列化器
         *  2:自定义反序列化器
         *  3:使用序列化器
 *  5：拦截器（不重要）

# 1:topic操作api #

## 1：pom.xml导包 ##

<dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>2.2.0</version>
            </dependency>
    
            
            <dependency>
                <groupId>log4j</groupId>
                <artifactId>log4j</artifactId>
                <version>1.2.17</version>
            </dependency>
            
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
                <version>1.7.25</version>
            </dependency>
            
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
                <version>1.7.25</version>
            </dependency>

## 2：创建topic ##

## 3：查看topic列表 ##

//配置连接参数
            Properties props = new Properties();
            props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
                        "CentOSA:9092,CentOSB:9092,CentOSC:9092");
            //可以用ip，前提是kafka服务器中server.properties中配置advertised.listeners=PLAINTEXT://192.168.138.128:9092
    //        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.138.128:9092");
    
            KafkaAdminClient adminClient= (KafkaAdminClient) KafkaAdminClient.create(props);
    //查询topics
            KafkaFuture<Set<String>> nameFutures = adminClient.listTopics().names();
            //加上get后就会同步
            for (String name : nameFutures.get()) {
        
                System.out.println(name);
            }

## 4：查看topic详情 ##

//配置连接参数
            Properties props = new Properties();
            props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
                        "CentOSA:9092,CentOSB:9092,CentOSC:9092");
            //可以用ip，前提是kafka服务器中server.properties中配置advertised.listeners=PLAINTEXT://192.168.138.128:9092
    //        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.138.128:9092");
    
            KafkaAdminClient adminClient= (KafkaAdminClient) KafkaAdminClient.create(props);
     //查看Topic详情
            DescribeTopicsResult describeTopics =
                    adminClient.describeTopics(Arrays.asList("topic01"));
            Map<String, TopicDescription> tdm = describeTopics.all().get();
            for (Map.Entry<String, TopicDescription> entry : tdm.entrySet()) {
        
                System.out.println(entry.getKey()+"\t"+entry.getValue());
            }

## 5：删除topic ##

# 2：生产者和消费者 #

## 1：生产者 ##

![在这里插入图片描述][de490f6843aa41128fd698a893a11b05.png]

### 1：普通异步发送 ###

public static void main(String[] args) throws InterruptedException {
        
            //1.创建链接参数
            Properties props=new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            //序列化key
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            //序列化value
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
    
            //2.创建生产者
            KafkaProducer<String,String> producer=new KafkaProducer<String, String>(props);
    
            //3.封账消息队列
            for(Integer i=0;i< 10;i++){
        
                Thread.sleep(100);
                ProducerRecord<String, String> record = new ProducerRecord<>("topic03", "key" + i, "value" + i);
                producer.send(record);
            }
    
            //关闭生产者
            producer.close();
        }

### 2：带回调函数的异步发送 ###

//带回调函数的异步发送
        public void test() throws InterruptedException {
        
            // 1. 创建 kafka 生产者的配置对象
            Properties properties = new Properties();
            // 2. 给 kafka 配置对象添加配置信息
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                    "hadoop102:9092");
            // key,value 序列化（必须）：key.serializer，value.serializer
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                    StringSerializer.class.getName());
    
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                    StringSerializer.class.getName());
            // 3. 创建 kafka 生产者对象
            KafkaProducer<String, String> kafkaProducer = new
                    KafkaProducer<String, String>(properties);
            // 4. 调用 send 方法,发送消息
            for (int i = 0; i < 5; i++) {
        
                // 添加回调
                kafkaProducer.send(new ProducerRecord<>("first",
                        "atguigu " + i), new Callback() {
        
    
                    // 该方法在 Producer 收到 ack 时调用，为异步调用
                    @Override
                    public void onCompletion(RecordMetadata metadata,
                                             Exception exception) {
        
                        if (exception == null) {
        
                            // 没有异常,输出信息到控制台
                            System.out.println(" 主题： " +
                                    metadata.topic() + "->" + "分区：" + metadata.partition());
                        } else {
        
                            // 出现异常打印
                            exception.printStackTrace();
                        }
                    }
                });
                // 延迟一会会看到数据发往不同分区
                Thread.sleep(2);
            }
            // 5. 关闭资源
            kafkaProducer.close();
        }

### 3:同步发送 ###

//同步发送
        public void test1() throws ExecutionException, InterruptedException {
        
            // 1. 创建 kafka 生产者的配置对象
            Properties properties = new Properties();
            // 2. 给 kafka 配置对象添加配置信息
    
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092 ");
            // key,value 序列化（必须）：key.serializer，value.serializer
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            // 3. 创建 kafka 生产者对象
            KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
            // 4. 调用 send 方法,发送消息
            for (int i = 0; i < 10; i++) {
        
                // 异步发送 默认
                // kafkaProducer.send(new   ProducerRecord<> ("first", "kafka" + i));
                // 同步发送
                kafkaProducer.send(new ProducerRecord<>("first", "kafka" + i)).get();
            }
            // 5. 关闭资源
            kafkaProducer.close();
        }

### 4:prop参数名称 描述 ###

*  bootstrap.servers 生产者连接集群所需的 broker 地 址 清 单 。 例 如hadoop102:9092,hadoop103:9092,hadoop104:9092，可以设置 1 个或者多个，中间用逗号隔开。注意这里并非需要所有的 broker 地址，因为生产者从给定的 broker里查找到其他 broker 信息。
 *  key.serializer 和 value.serializer 指定发送消息的 key 和 value 的序列化类型。一定要写全类名。
 *  buffer.memory RecordAccumulator 缓冲区总大小，默认 32m。
 *  batch.size 缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
 *  linger.ms 如果数据迟迟未达到 batch.size，sender 等待 linger.time之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。
 *  acks 0：生产者发送过来的数据，不需要等数据落盘应答。
 *  1：生产者发送过来的数据，Leader 收到数据后应答。
 *  \-1（all）：生产者发送过来的数据，Leader+和 isr 队列里面的所有节点收齐数据后应答。默认值是-1，-1 和all 是等价的。
 *  max.in.flight.requests.per.connection 允许最多没有返回 ack 的次数，默认为 5，开启幂等性  
    要保证该值是 1-5 的数字。
 *  retries 当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是 int 最大值，2147483647。如果设置了重试，还想保证消息的有序性，需要设置MAX\_IN\_FLIGHT\_REQUESTS\_PER\_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。
 *  retry.backoff.ms 两次重试之间的时间间隔，默认是 100ms。
 *  enable.idempotence 是否开启幂等性，默认 true，开启幂等性。
 *  compression.type 生产者发送的所有数据的压缩方式。默认是 none，也  
    就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd。

## 2：消费者 ##

![在这里插入图片描述][51f0ded733da43389b987ed262d0bf1d.png]

### 1:消费者组初始化流程 ###

![在这里插入图片描述][74e241f6387d435d88f23e65718e1d48.png]  
![在这里插入图片描述][0d4761d6a4d24edabe1d1b22476af258.png]

### 2：参数名称 描述 ###

*  bootstrap.servers 向 Kafka 集群建立初始连接用到的 host/port 列表。  
    \*key.deserializer 和value.deserializer指定接收消息的 key 和 value 的反序列化类型。一定要写全  
    类名。
 *  group.id 标记消费者所属的消费者组。
 *  enable.auto.commit 默认值为 true，消费者会自动周期性地向服务器提交偏移量。
 *  auto.commit.interval.ms 如果设置了 enable.auto.commit 的值为 true， 则该值定义了消费者偏移量向 Kafka 提交的频率，默认 5s。
 *  auto.offset.reset 当 Kafka 中没有初始偏移量或当前偏移量在服务器中不存在（如，数据被删除了），该如何处理？ earliest：自动重置偏移量到最早的偏移量。 latest：默认，自动重置偏移量为最  
    新的偏移量。 none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常。 anything：向消费者抛异常。
 *  offsets.topic.num.partitions \_\_consumer\_offsets 的分区数，默认是 50 个分区。
 *  heartbeat.interval.ms Kafka 消费者和 coordinator 之间的心跳时间，默认 3s。该条目的值必须小于 session.timeout.ms ，也不应该高于session.timeout.ms 的 1/3。
 *  session.timeout.ms Kafka 消费者和 coordinator 之间连接超时时间，默认 45s。超过该值，该消费者被移除，消费者组执行再平衡。
 *  max.poll.interval.ms 消费者处理消息的最大时长，默认是 5 分钟。超过该值，该  
    消费者被移除，消费者组执行再平衡。
 *  fetch.min.bytes 默认 1 个字节。消费者获取服务器端一批消息最小的字节  
    数。
 *  fetch.max.wait.ms 默认 500ms。如果没有从服务器端获取到一批数据的最小字  
    节数。该时间到，仍然会返回数据。
 *  fetch.max.bytes 默认 Default: 52428800（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 message.max.bytes （broker config）or max.message.bytes （topic config）影响。
 *  max.poll.records 一次 poll 拉取数据返回消息的最大条数，默认是 500 条。

### 3:订阅主题 ###

![在这里插入图片描述][f32e856460cc46999af7b9f50eda1ae3.png]

public static void main(String[] args) {
            //1.创建Kafka链接参数
            Properties props=new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
            props.put(ConsumerConfig.GROUP_ID_CONFIG,"group01");
    
            //2.创建Topic消费者
            KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(props);
            //3.订阅topic开头的消息队列
            //这里可以传正则还可以直接传list
            consumer.subscribe(Pattern.compile("^topic.*$"));
    
            while (true){
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
                Iterator<ConsumerRecord<String, String>> recordIterator = consumerRecords.iterator();
                while (recordIterator.hasNext()){
                    ConsumerRecord<String, String> record = recordIterator.next();
                    String key = record.key();
                    String value = record.value();
                    long offset = record.offset();
                    int partition = record.partition();
                    System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset);
                }
            }
        }

### 4：订阅分区 ###

![在这里插入图片描述][5878176b4cb34227bca71e269ed8c14c.png]

public void test(){
        
            Properties properties = new Properties();
    
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
            // 配置序列化 必须
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                    StringDeserializer.class.getName());
    
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                    StringDeserializer.class.getName());
            // 配置消费者组（必须），名字可以任意起
            properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
            KafkaConsumer<String, String> kafkaConsumer = new
                    KafkaConsumer<>(properties);
            // 消费某个主题的某个分区数据
            ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
            topicPartitions.add(new TopicPartition("first", 0));
            kafkaConsumer.assign(topicPartitions);
            while (true){
        
                ConsumerRecords<String, String> consumerRecords =
                        kafkaConsumer.poll(Duration.ofSeconds(1));
                for (ConsumerRecord<String, String> consumerRecord :
                        consumerRecords) {
        
                    System.out.println(consumerRecord);
                }
            }
        }

# 3：自定义分区 #

## 1：系统默认分区策略 ##

生产者在发送消息时，可以将消息发往不同的分区，我们系统默认的分区策略（DefaultPartitioner。）如下：  
![9259afc976c3423f82f29b43b343ca2f.png][]

## 2：自定义分区策略 ##

我们如果有需求，也可以自定义分区策略

### 1：分区策略默认实现Partitioner ###

package com.kafka.partitioner;
    
    import org.apache.kafka.clients.producer.Partitioner;
    import org.apache.kafka.common.Cluster;
    import org.apache.kafka.common.utils.Utils;
    
    import java.util.Map;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class UserDefinePartitioner  implements Partitioner {
        
        private AtomicInteger atomicInteger=new AtomicInteger(0);
        @Override
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        
            int numPartitions = cluster.partitionsForTopic(topic).size();
            if(keyBytes==null || keyBytes.length==0){
        
                return atomicInteger.addAndGet(1) & Integer.MAX_VALUE% numPartitions;
            } else {
        
                return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
            }
        }
    
        @Override
        public void close() {
        
            System.out.println("close");
        }
    
        @Override
        public void configure(Map<String, ?> configs) {
        
            System.out.println("configure");
        }
    }

### 2:使用自定义分区策略 ###

package com.kafka.partitioner;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    public class KafkaProducerDemo {
        
        public static void main(String[] args) {
        
            //1.创建链接参数
            Properties props=new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
            props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,UserDefinePartitioner.class.getName());
    
            //2.创建生产者
            KafkaProducer<String,String> producer=new KafkaProducer<String, String>(props);
    
            //3.封账消息队列
            for(Integer i=0;i< 10;i++){
        
                ProducerRecord<String, String> record = new ProducerRecord<>("topic01",  "value" + i);
                producer.send(record);
            }
    
            producer.close();
        }
    }

# 4:序列化（重要） #

因为kafka也是二进制安全的，即只接受二级制数据。所以我们传入的key和value都需要序列化

## 1：系统默认的序列化器 ##

系统默认对所有基本类型都进行了序列化  
![在这里插入图片描述][7399991eacc0409cbee8ceddbe5a5252.png]  
以及反序列化器  
![在这里插入图片描述][6fcad97075b348a0894b99802da154ee.png]

## 2：自定义序列化 ##

### 1：自定义序列化器 ###

package com.kafka.serializer;
    
    import org.apache.commons.lang3.SerializationUtils;
    import org.apache.kafka.common.serialization.Serializer;
    
    import java.io.Serializable;
    import java.util.Map;
    
    public class ObjectSerializer implements Serializer<Object> {
        
    
        @Override
        public void configure(Map<String, ?> configs, boolean isKey) {
        
            System.out.println("configure");
        }
    
        @Override
        public byte[] serialize(String topic, Object data) {
        
            return SerializationUtils.serialize((Serializable) data);
        }
    
        @Override
        public void close() {
        
            System.out.println("close");
        }
    }

### 2:自定义反序列化器 ###

package com.kafka.serializer;
    
    import org.apache.commons.lang3.SerializationUtils;
    import org.apache.kafka.common.serialization.Deserializer;
    
    import java.util.Map;
    
    public class ObjectDeserializer implements Deserializer<Object> {
        
        @Override
        public void configure(Map<String, ?> configs, boolean isKey) {
        
            System.out.println("configure");
        }
    
        @Override
        public Object deserialize(String topic, byte[] data) {
        
            return SerializationUtils.deserialize(data);
        }
    
        @Override
        public void close() {
        
            System.out.println("close");
        }
    }

### 3:使用序列化器 ###

package com.kafka.serializer;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.util.Date;
    import java.util.Properties;
    
    public class KafkaProducerDemo {
        
        public static void main(String[] args) {
        
            //1.创建链接参数
            Properties props=new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,ObjectSerializer.class.getName());
    
            //2.创建生产者
            KafkaProducer<String,User> producer=new KafkaProducer<String, User>(props);
    
            //3.封账消息队列
            for(Integer i=0;i< 10;i++){
        
                ProducerRecord<String, User> record = new ProducerRecord<>("topic01", "key"+i,new User(i,"user"+i,new Date()));
                producer.send(record);
            }
    
            producer.close();
        }
    }

# 5：拦截器（不重要） #

拦截器可以对发送的消息进行处理，也可以对返回的消息ack等进行处理  
1：自定义拦截器

package com.kafka.interceptors;
    
    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 UserDefineProducerInterceptor implements ProducerInterceptor {
        
        @Override
        public ProducerRecord onSend(ProducerRecord record) {
        
            ProducerRecord wrapRecord = new ProducerRecord(record.topic(), record.key(), record.value());
            wrapRecord.headers().add("user","mashibing".getBytes());
            return wrapRecord;
    
        }
    
        @Override
        public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        
            System.out.println("metadata:"+metadata+",exception:"+exception);
        }
    
        @Override
        public void close() {
        
            System.out.println("close");
        }
    
        @Override
        public void configure(Map<String, ?> configs) {
        
            System.out.println("configure");
        }
    }

[de490f6843aa41128fd698a893a11b05.png]: https://img-blog.csdnimg.cn/de490f6843aa41128fd698a893a11b05.png
[51f0ded733da43389b987ed262d0bf1d.png]: https://img-blog.csdnimg.cn/51f0ded733da43389b987ed262d0bf1d.png
[74e241f6387d435d88f23e65718e1d48.png]: https://img-blog.csdnimg.cn/74e241f6387d435d88f23e65718e1d48.png
[0d4761d6a4d24edabe1d1b22476af258.png]: https://img-blog.csdnimg.cn/0d4761d6a4d24edabe1d1b22476af258.png
[f32e856460cc46999af7b9f50eda1ae3.png]: https://img-blog.csdnimg.cn/f32e856460cc46999af7b9f50eda1ae3.png
[5878176b4cb34227bca71e269ed8c14c.png]: https://img-blog.csdnimg.cn/5878176b4cb34227bca71e269ed8c14c.png
[9259afc976c3423f82f29b43b343ca2f.png]: https://img-blog.csdnimg.cn/9259afc976c3423f82f29b43b343ca2f.png
[7399991eacc0409cbee8ceddbe5a5252.png]: https://img-blog.csdnimg.cn/7399991eacc0409cbee8ceddbe5a5252.png
[6fcad97075b348a0894b99802da154ee.png]: https://img-blog.csdnimg.cn/6fcad97075b348a0894b99802da154ee.png