【Kafka】Apache Kafka消息队列JavaAPI实战

本是古典何须时尚 2023-10-05 12:26 22阅读 0赞

#### 指路牌 ####

*   *  Kafka API实战
     *   *  依赖
         *  log4j配置
         *  基础篇
         *   *  消费者
             *  生产者
             *  序列化
             *  自定义分区
             *  拦截器
         *  高级篇
         *   *  Offset控制
             *  Acks&Retries
             *  幂等性
             *  事务控制

### Kafka API实战 ###

#### 依赖 ####

<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>

#### log4j配置 ####

log4j.rootLogger = info,console
    
    log4j.appender.console = org.apache.log4j.ConsoleAppender
    log4j.appender.console.Target = System.out
    log4j.appender.console.layout = org.apache.log4j.PatternLayout
    log4j.appender.console.layout.ConversionPattern =  %p %d{yyyy-MM-dd HH:mm:ss} %c - %m%n

#### 基础篇 ####

##### 消费者 #####

Properties props = new Properties();
    //必须配置
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOS:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG,"g3");
    //默认配置
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    //如果消费者没有订阅过消息，默认会从当前最新的offset位置开始消费，如果用户希望从最早位置消费可以配置为earliest，默认值是latest
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
    
    Consumer<String,String> consumer= new KafkaConsumer<String,String>(props);
    //订阅相关的topic开头的所有消息
    consumer.subscribe(Pattern.compile("^topic.*")); //consumer.subscribe(Arrays.asList("topic01"));
    
    //迭代遍历消息队列
    try {
       
        
        while (true){
       
        
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));//抓取数据，指定抓取的周期
            if(records!=null && !records.isEmpty()){
       
        //迭代结果
                Iterator<ConsumerRecord<String, String>> recordIterator = records.iterator();
                while (recordIterator.hasNext()){
       
        
                    ConsumerRecord<String, String> record = recordIterator.next();
                    System.out.println(record.key()+"\t"+record.value()+"\t"+record.offset()+"\t"+record.partition()+"\t"+record.timestamp());
                }
            }
        }
    } catch (Exception e) {
       
        
        consumer.close();
    }

##### 生产者 #####

Properties props = new Properties();
    //必选参数
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOS:9092");
    //默认配置 可选
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
    //优化配置 批处理、缓冲
    props.put(ProducerConfig.BATCH_SIZE_CONFIG,1000);//设置分区 批次大小1000 bytes
    props.put(ProducerConfig.LINGER_MS_CONFIG,1000);//最多等待1s钟
    
    Producer<String,String> producer= new KafkaProducer<String,String>(props);
    
    for (int i = 0; i < 10; i++) {
       
        
        DecimalFormat format = new DecimalFormat("00");
        String key = format.format(i);
        //默认会按照 hash（key）% 分区数
        ProducerRecord<String,String> record=new ProducerRecord<String, String>("topic01",key,"value"+ key);
        producer.send(record);
    }
    producer.flush();
    producer.close();

##### 序列化 #####

*反序列化*

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"<