Kafka之自定义生产者与消费者

叁歲伎倆 2022-12-28 14:15 252阅读 0赞

# Kafka之自定义生产者与消费者 #

## 一、自定义生产者代码 ##

package com.kafka.product;
    
    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.protocol.types.Field;
    
    import java.util.Properties;
    import java.util.concurrent.ExecutionException;
    
    public class MyProduct { 
    
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
    
            //1.创建producer client
            Properties props = new Properties();
            //指定key的序列化器
            props.setProperty("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
            //指定value的序列化器
            props.setProperty("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
    
            //指定ack
            props.setProperty("acks","-1");
            //设置一个批次的大小
            props.setProperty("batch.size", "1024");
    
            //指定kafka集群
            props.setProperty("bootstrap.servers","hadoop102:9092,hadoop103:9092");
    
            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
    
            //2.封装数据
            for (int i=701;i<=800;i++){ 
    
                String msg = "这是第"+i+"条数据";
    
                final int num = i;
    
                ProducerRecord<String, String> record = new ProducerRecord<String, String>("second", msg);
    
                System.out.println("开始发送第"+i+"条数据");
    
                //3.进行数据发送
                //异步写入
                producer.send(record, new Callback() { 
                    /** * kafka返回确认消息的时候调用 * @param metadata * @param e */
                    public void onCompletion(RecordMetadata metadata, Exception e) { 
    
                        System.out.println("这是第"+num+"条数据已经发送到了kafka");
    
                        String topic = metadata.topic();
                        int partition = metadata.partition();
                        long offset = metadata.offset();
    
                        System.out.println("topic="+topic+",partition="+partition+",offset="+offset);
    
                    }
                    //同步发送数据
                }).get();
    
                System.out.println("第"+i+"条数据发送完成");
    
            }
    
            //4.关闭
            producer.close();
    
    
        }
    
    }

## 二、自定义消费者代码 ##

package com.kafka.consumer;
    
    import org.apache.kafka.clients.consumer.*;
    import org.apache.kafka.common.TopicPartition;
    
    import java.time.Duration;
    import java.util.*;
    
    public class MyConsumer { 
    
        public static void main(String[] args) { 
    
            //1.创建consumer client
            Properties prop = new Properties();
            //设置key的反序列化器
            prop.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
            //设置value的反序列化器
            prop.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
            //指定kafka集群
            prop.setProperty("bootstrap.servers","hadoop102:9092,hadoop103:9092");
            //设置消费者组的名称
            prop.setProperty("group.id","cc1");
    
            //设置消费者组第一次消费时从哪里开始消费
            prop.setProperty("auto.offset.reset","earliest");
    
            //设置是否自动提交offset
            prop.setProperty("enable.auto.commit","false");
    
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(prop);
    
    
            //2.执行数据获取
            while (true){ 
                //设置从哪些topic中获取数据
                List<String> topics = new ArrayList<String>();
                topics.add("second");
                consumer.subscribe(topics);
    
                //设置间隔多久拉取一次数据
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
    
                Iterator<ConsumerRecord<String, String>> it = records.iterator();
    
                while (it.hasNext()){ 
    
                    ConsumerRecord<String, String> msg = it.next();
                    String topic = msg.topic();
                    int partition = msg.partition();
                    String value = msg.value();
                    long offset = msg.offset();
    
                    System.out.println("topic="+topic+",partition="+partition+",value="+value+",offset="+offset);
    
                }
    
                //手动提交offset
                consumer.commitAsync(new OffsetCommitCallback() { 
    
                    /** * 提交完成调用 * */
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) { 
    
                        Collection<OffsetAndMetadata> values = offsets.values();
    
                        for (OffsetAndMetadata metadata : values) { 
                            long offset = metadata.offset();
                            System.out.println("提交数据的offset="+offset);
                        }
    
                    }
                });
    
            }
    
        }
    
    }