一文读懂kafka

查看有多少个topic

./kafka-topics.sh --bootstrap-server 10.10.5.252:9092   --list

创建topic

./kafka-topics.sh --bootstrap-server 10.10.5.252:9092  --topic windaka_One e --create --partitions 1 --replication-factor 2

查看主题信息

./kafka-topics.sh --bootstrap-server 10.10.5.252:9092 --topic windaka_One --describe

增加分区

./kafka-topics.sh --bootstrap-server 10.10.5.250:9092 --topic windaka_One --alter --partitions 3

生产者消费者发送接收信息

./kafka-console-producer.sh --bootstrap-server 10.10.5.252:9092 --topic windaka_One

./kafka-console-consumer.sh --bootstrap-server 10.10.5.252:9092 --topic windaka_One

异步发送


org.apache.kafka
kafka-clients
3.0.0

public class CustomProducer{

    public static void main(String[] args){

        
        Properties properties  = new Properties();

        
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.10.5.252:9092,10.10.5.250:9092");

        
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        
        kafkaProducer producer_0 = new kafkaProducer(properties);

        
        producer_0.send(new ProducerRecord<>("windaka_One(主题)“,"消息内容 Test-kafka"));

        
        producer_0.close();
    }

}

带回调函数的异步发送

producer_0.send(new ProducerRecord<>("windaka_One主题","发送数据内容"),
       new Callback(){
        
          @Override
          public void onCompletion(RecordMetadata record,Exception e){
              if(e == null){
                  System.out.println("Topic:"+record.topic());
                  System.out.println("分区:"+record.partition());
              }
          }
       });

同步发送

producer_0.send(new ProducerRecord<>("主题","发送数据内容")).get();

自定义分区器实现一个类继承Partitioner

public class MyPartitioner implements Partitioner{

    @Override
    public int partition(String topic,Object o,byte[] bytes,Object value,byte[] bytes1,Cluster cluster){
        String msgValue = value.toString();
        int partition;
        if(msgValue.contains("windaka")){
            partition = 0;
        }else{
            partition = 1;
        }
        return partition;
    }


}

属性设置里面进行关联自定义分区设置


properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.kafka.test.MyPartitioner");

生产者如何提高吞吐量

properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554422);


properties.put(ProducerConfig.BATCH_SIZE_FONGI,16384);


properties.put(ProducerConfig.LINGER_MS_CONFIG,5);



properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");

生产经验-数据可靠性

properties.put(ProducerConfig.ACKS_CONFIG,"1");


properties.put(ProducerConfig.RETRIES_CONFIG,3);

数据去重-正好一次

public class CustomProducer{

    public static void main(String[] args){
        
        
        Properties properties = new Properties();

        
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.10.5.252:9092,10.10.5.250:9092");

        
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

     
      properties.put(ProducerConfig.ACKS_CONFIG,"-1");

     
      properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"windaka01");

      
      KafkaProducer producer_0 = new KafkaProducer(properties);

      
      producer_0.initTransactions();

      
      producer_0.beginTransaction();

      try{
          producer_0.send(new ProducerRecord<>("topic","发送数据内容"));

          
          producer_0.commitTransaction();
      }catch(Exception e){
          
          producer_0.abortTransaction();
      }finally{
          
          producer_0.close();
      }   
    }
}

生产经验-数据有序
单分区内有序

多分区，分区与分区间无序

Zookeeper常用命令
bin/zkCli.sh

增加服役新节点，负载均衡
（1）创建一个要均衡的主题

（3）创建执行副本

vi increase-replication-factor.json

把第二步生成的均衡计划复制到执行副本里面。

退役旧节点

（6）.停止退役节点上kafka服务即可。

kafka副本

Leader Partition自动平衡

Ps:

建议不使用再平衡。

增加副本

kafka文件存储机制

命令：
kafka-run-class.sh kafka.tools.DumpLogSegments --files ./0000000000000000.log

日志清理策略

kafka高效读写数据原理

消费者消费一个主题数据

public class CustomConsumer{

    public static void main(String[] args){

        
        Properties pro = new Properties();

        
        pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.10.5.252:9092,10.10.5.250:9092");
        
        pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
       
       
       pro.put(ConsumerConfig.GROUP_ID_CONFIG,"test消费者group id");
       
       
       KafkaConsumer consumer = new KafkaConsumer(pro);

       
       ArrayList topic = new ArrayList<>();
       
       topic.add("windaka_One");
       consumer.subscribe(topic);

       
       while(true){
           
           ConsumerRecords conRecords = consumer.poll(Duration.ofseconds(1));

           for(ConsumerRecord record : conRecords){
                System.out.println(record.topic());
                System.out.println(record.value());
                System.out.println(record.partition());
           }
       }

    }

}

消费一个主题下的某个分区数据

public class CustomConsumer{

    public static void main(String[] args){

        
        Properties pro = new Properties();

        
        pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.10.5.252:9092,10.10.5.250:9092");
        
        
        pro.put(ConsumerConfig.GROUP_ID_CONFIG,"test");

        
        pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserialize.class.getName());
        
        
        KafkaConsumer consumer = new KafkaConsumer(pro);

        
        ArrayList topicPart = new ArrayList<>();
        
        topicPart.add(new TopicPartition("windaka_One",0));
        consumer.assign(topicPart);

        
        while(true){
            ConsumerRecords conRecords = consumer.poll(Duration.ofSeconds(1));

            for(ConsumerRecord record : conRecords){
                System.out.println(record.topic());
                System.out.println(record.value());
                System.out.println(record.partition());
            }
        }
    }

}

消费者分区策略
-->Range

-->RoundRobin

pro.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");

-->Sticky

pro.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");

消费者offset
0.9版本之前维护在zookeep中；

0.9版本之后维护在消费者主题当中；

自动offset提交

pro.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);


pro.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);

手动提交offset

--同步提交

public class CustomConsumer{

    public static void main(String[] args){

        Properties pro = new Properties();

        pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.10.5.252:9092,10.10.5.250:9092");

        pro.put(ConsumerConfig.GROUP_ID_CONFIG,"test");

        pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());

        
        pro.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

        KafkaConsumer consumer = new KafkaConsumer(pro);

        ArrayList topic = new ArrayList<>();
        topic.add("windaka_One");
        consumer.subscribe(topic);

        while(true){
            ConsumerRecords conRecords = consumer.poll(Duration.ofSeconds(1));

            for(ConsumerRecord record : conRecords){
                System.out.println(record.topic());
                System.out.println(record.value());
            }

            
            consumer.commitSync();
        }
    }

}

--异步提交

public class CustomConsumer{

    public static void main(String[] args){

        Properties pro = new Properties();

        pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.10.5.252:9092,10.10.5.250:9092");

        pro.put(ConsumerConfig.GROUP_ID_CONFIG,"test");

        pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());

        
        pro.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

        KafkaConsumer consumer = new KafkaConsumer(pro);

        ArrayList topic = new ArrayList<>();
        topic.add("windaka_One");
        consumer.subscribe(topic);

        while(true){
            ConsumerRecords conRecords = consumer.poll(Duration.ofSeconds(1));

            for(ConsumerRecord record : conRecords){
                System.out.println(record.topic());
                System.out.println(record.value());
            }

            
            consumer.commitAsync();
        }
    }

}

消费者指定offset进行消费

KafkaConsumer consumer = new KafkaConsumer(pro);

ArrayList  topic = new ArrayList<>();
topic.add("windaka_One");
consumer.subscribe(topic);


Set  assignment = consumer.assignment();


while(assignment.size() == 0){
    consumer.poll(Duration.ofSeconds(1));
    assignment = consumer.assignment();
}

for(TopicPartition topicPart : assignment){
    
    consumer.seek(topicPart,100);
}

指定消费时间进行消费

Set  assignment = consumer.assignment();


while(assignment.size() == 0){
    consumer.poll(Duration.ofSeconds(1));
    assignment = consumer.assignment();
}


HashMap  topicPartHsM = new HashMap<>();

for(TopicPartition topicPartition : assignment){
    topicPartHsm.put(topicPartition,System.currentTimeMillis()*1*24*3600*1000);
}

Map topicPartOfHsm = consumer.offsetForTimes(topicPartHsM);

for(TopicPartition topicPart : assignment){
    OffsetAndTimestamp offsetTimeStamp = topicPartOfHSM.get(topicPart);
    consumer.seek(topicPart,offsetTimeStamp.offset());
}

数据积压-消费者如何提供吞吐量

EFAK(eagle) EFAK原名为eagle,是kafka的开源可视化监控系统。

Kafka-Kraft模式 2.8.0版本以后新模式，目的是为了去zookeeper化
（不安装启动zookeeper，一样使用kafka）

conf/kraft/server.properties
process.roles = broker,controller


node.id = 252


controller.quorum.voters=252@10.10.5.252:9093,250@10.10.5.250:9093

advertised.listeners=PLANTEXT://10.10.5.252:9092

log.dirs=/xxx/xxx/xxx

外部组件 kafka集成Flume
1.下载解压Flume 2. 删除不兼容包
rm -rf guava-11.0.2.jar
3.安装配置hadoop 4.vi log4j.properties
配置log目录
5.配置Flume
a1.sources.r1.filegroups.f1 = /var/log/xxx (监控的文件)
6. 启动Flume

外部组件 kafka集成Flink

集成方法不再记述

外部组件 kafka集成SpringBoot

一文读懂kafka

Java相关栏目本月热门文章