Kafka(四).Kafka&JAVA 基础API

Kafka(四).Kafka&JAVA 基础API 1.环境

使用java 来测试Kafka API 运行环境基于Kafka(三)的搭建的集群环境；

测试电脑(windows) 需要配置host

192.168.141.131 CentOSA
192.168.141.132 CentOSB
192.168.141.133 CentOSC

maven配置

		
            org.apache.kafka
            kafka-clients
            2.7.0
        

        
            log4j
            log4j
            1.2.17
        

        
            org.slf4j
            slf4j-log4j12
            1.7.7
        

        
            org.slf4j
            slf4j-api
            1.7.7

2. 查看&创建&删除&查看 topic

public class KafkaTopicDML {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1.创建KafkaAdminClient  相当创建了一个topic 连接的配置
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");

        KafkaAdminClient adminClient = (KafkaAdminClient)KafkaAdminClient.create(properties);

        ListTopicsResult listTopics = adminClient.listTopics();
        Set names = listTopics.names().get();
        for (String name : names) {
            System.out.println("topic 为："+ name);
        }
        
        //异步创建  
        adminClient.createTopics(Arrays.asList(new NewTopic("topic03",3,(short)3)));
        
	    //同步创建
        //CreateTopicsResult topic05 = adminClient.createTopics(Arrays.asList(new NewTopic("topic05", 3, (short) 3)));
        //topic05.all().get();
        
        //异步删除
        adminClient.deleteTopics(Arrays.asList("topic05", "topic06"));
        
        //同步删除
        //DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList("topic05", "topic06"));
        //deleteTopicsResult.all().get();
        
        //同步查看
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(names);
        Map stringTopicDescriptionMap = describeTopicsResult.all().get();
        System.out.println(stringTopicDescriptionMap);
        
        
        //关闭 adminClient
        adminClient.close();
    }
}

3. 生产者& 消费者

//生产者
public class KafkaProducerDemo {

    public static void main(String[] args) {
        //1.创建 KafkaProducer
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        KafkaProducer kafkaProducer = new KafkaProducer (properties);
        for (int i = 0; i < 10; i++) {
            ProducerRecord record = new ProducerRecord<>("topic02", "key" + i, "value" + i);
            //发送
            kafkaProducer.send(record);
        }

        kafkaProducer.close();
    }
}

//消费者
public class KafkaConsumerDemo {

    public static void main(String[] args) {
        //1.创建 KafkaProducer
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC: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,"g2");

        KafkaConsumer kafkaConsumer = new KafkaConsumer (properties);
        kafkaConsumer.subscribe(Pattern.compile("^topic02.*"));
        while (true){
            ConsumerRecords records = kafkaConsumer.poll(Duration.ofSeconds(1));
            if(!records.isEmpty()){
                Iterator> iterator = records.iterator();
                while (iterator.hasNext()){
                    ConsumerRecord next = iterator.next();
                    System.out.println(next);
                }
            }
        }
    }
}

测试验证不指定分区的情况下

###打开一个消费者   消费了三个分区  S1
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Notifying assignor about the new Assignment(partitions=[topic02-0, topic02-1, topic02-2])

###再打开一个消费者  S2  ====>S1的一个分区被同组另一个消费者S2获取了
##S1 log
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Finished assignment for group at generation 8: {consumer-g2-1-58d640e4-0269-4c1c-a212-5f21bc761cca=Assignment(partitions=[topic02-0, topic02-1]), consumer-g2-1-81b66c7f-6c84-4095-809c-ce7afb4100a9=Assignment(partitions=[topic02-2])}

=======================================================================

##S2 log
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Notifying assignor about the new Assignment(partitions=[topic02-2])

####再打开一个消费者  S3  ====>====>S1的一个分区被同组另一个消费者S3获取了
##s1 log
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Finished assignment for group at generation 9: {consumer-g2-1-27f9cab4-cacc-425c-81c8-6e9fc141580c=Assignment(partitions=[topic02-0]), consumer-g2-1-58d640e4-0269-4c1c-a212-5f21bc761cca=Assignment(partitions=[topic02-1]), consumer-g2-1-81b66c7f-6c84-4095-809c-ce7afb4100a9=Assignment(partitions=[topic02-2])}

###S3 log
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Notifying assignor about the new Assignment(partitions=[topic02-0])

=======================================================================

###再打开一个消费者  S4  ====>已经没有分区在分给新增的同组消费者了
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Notifying assignor about the new Assignment(partitions=[])

=======================================================================
##关闭S2   候补S4 接替S2监听的分区
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Notifying assignor about the new Assignment(partitions=[])
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Adding newly assigned partitions: 
INFO  - AbstractCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Attempt to heartbeat failed since group is rebalancing
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Revoke previously assigned partitions 
INFO  - AbstractCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] (Re-)joining group
INFO  - AbstractCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Successfully joined group with generation Generation{generationId=11, memberId='consumer-g2-1-8f55c39e-9bf4-45a5-a35b-42ab9eb64d4f', protocol='range'}
INFO  - AbstractCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Successfully synced group in generation Generation{generationId=11, memberId='consumer-g2-1-8f55c39e-9bf4-45a5-a35b-42ab9eb64d4f', protocol='range'}
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Notifying assignor about the new Assignment(partitions=[topic02-2])
INFO  - ConsumerCoordinator        - [Consumer clientId=consumer-g2-1, groupId=g2] Adding newly assigned partitions: topic02-2

得出：消费者组内消费者每添加一个，消费者组内部会有一个简单的负载均衡机制，当消费者组成员数目大于分区数的时候就没有分区可以分配了，会进行候补；直到其中一个消费者宕机，最后新增的才可以替补宕机消费者的分区；

当开启三个消费者S1 S2 S3时开启生产者生产10个记录

=======================================================================
##S1
ConsumerRecord(topic = topic02, partition = 1, leaderEpoch = 2, offset = 7, CreateTime = 1632753328402, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key0, value = value0)
ConsumerRecord(topic = topic02, partition = 1, leaderEpoch = 2, offset = 8, CreateTime = 1632753328416, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key3, value = value3)
ConsumerRecord(topic = topic02, partition = 1, leaderEpoch = 2, offset = 9, CreateTime = 1632753328416, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key5, value = value5)
ConsumerRecord(topic = topic02, partition = 1, leaderEpoch = 2, offset = 10, CreateTime = 1632753328416, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key8, value = value8)

=======================================================================
##S2
ConsumerRecord(topic = topic02, partition = 2, leaderEpoch = 2, offset = 5, CreateTime = 1632753328415, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key1, value = value1)
ConsumerRecord(topic = topic02, partition = 2, leaderEpoch = 2, offset = 6, CreateTime = 1632753328416, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key2, value = value2)
ConsumerRecord(topic = topic02, partition = 2, leaderEpoch = 2, offset = 7, CreateTime = 1632753328416, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key9, value = value9)

=======================================================================
##S3
ConsumerRecord(topic = topic02, partition = 0, leaderEpoch = 0, offset = 6, CreateTime = 1632753328416, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key4, value = value4)
ConsumerRecord(topic = topic02, partition = 0, leaderEpoch = 0, offset = 7, CreateTime = 1632753328416, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key6, value = value6)
ConsumerRecord(topic = topic02, partition = 0, leaderEpoch = 0, offset = 8, CreateTime = 1632753328416, serialized key size = 4, serialized value size = 6, headers = RecordHeaders(headers = [], isReadonly = false), key = key7, value = value7)

得出：消费者消费的数据实在分区内有序在分区之间比较是没有顺序的

4.指定消费那个分区

可以指定消费开始的偏移量失去消费组的特性，消费者实例之间没有任何关系了

public class KafkaConsumerDemo_1 {
	//同时启动多个实例  实例之间是没有联系的
    public static void main(String[] args) {
        //1.创建 KafkaProducer
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC: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,"g2");  不需要指定组了

        KafkaConsumer kafkaConsumer = new KafkaConsumer (properties);
        List topics = Arrays.asList(new TopicPartition("topic02", 0));
        //指定订阅那个topic  那个分区 Partition
        kafkaConsumer.assign(topics);
        kafkaConsumer.seek(new TopicPartition("topic02", 0),3);//还可以设置偏移量  

        while (true){
            ConsumerRecords records = kafkaConsumer.poll(Duration.ofSeconds(1));
            if(!records.isEmpty()){
                Iterator> iterator = records.iterator();
                while (iterator.hasNext()){
                    ConsumerRecord next = iterator.next();
                    System.out.println(next);
                }
            }
        }
    }
}

4.1负载均衡

当启动三个消费者是发送30 条数据得到的log ===>默认的不是轮训而是hash

##S1  ==》11
key4
key6
key7
key10
key13
key14
key23
key24
key25
key26
key29

##S2  ==》9
key1
key2
key9
key11
key12
key19
key20
key21
key22
##S3   ==》10
key0
key3
key5
key8
key15
key16
key17
key18
key27
key28

4.2编写简单轮训

可以通过实现Partitioner接口并配置到KafkaProducer中来实现指定分区发送数据

//指定生产者分区 

public class KafkaProducerDemo_2 {

    public static void main(String[] args) {
        //1.创建 KafkaProducer
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "CentOSA:9092,CentOSB:9092,CentOSC:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
        KafkaProducer kafkaProducer = new KafkaProducer(properties);
        for (int i = 0; i < 10; i++) {
            ProducerRecord record = new ProducerRecord<>("topic02", "key" + i, "value" + i);
            kafkaProducer.send(record);
        }
        kafkaProducer.close();
    }

}
//MyPartitioner 是实现了轮训的   分区策略类  需要实现Partitioner 接口
public class MyPartitioner implements Partitioner {
    public MyPartitioner() {
    }
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        int size = cluster.topics().size();
        Integer key1 = null;
        try {
            key1 = Integer.valueOf(key.toString().replace("key", ""));
        } catch (NumberFormatException e) {
             return 1;
        }
        return key1%3;
    }
    @Override
    public void close() {
    }
    @Override
    public void configure(Map configs) {
    }
}

5.发送对象序列化

可以使用现成的序列化包

		 
            org.apache.commons
            commons-lang3
            3.9

配置对象序列化

public class ObjectSerializer implements Serializer {
    @Override
    public byte[] serialize(String topic, Object data) {
        return SerializationUtils.serialize((Serializable) data);
    }
}

配置对象反序列化

public class ObjectDeserializer implements Deserializer

Kafka(四).Kafka&JAVA 基础API

Java相关栏目本月热门文章