美文网首页
kafka learning note 2

kafka learning note 2

作者: 简为2016 | 来源:发表于2020-03-08 16:11 被阅读0次

    Consumer Group

    • consumers join in the group with same group id.
    • the maximum parallelism is the number of consumers not partitions.
    • Kafka allocates partitions of the topic to the users in the group, to ensure that one user, one partition.
    • Kafka ensures message only read by one consumer in the group.
    • consumers can check the message in sequence of stored message in logs.

    rebalancd consumers

    import java.util.Properties;
    import java.util.Arrays;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    
    public class ConsumerGroup{
            public static void main(String[] args)throws Exception{
                    if(args.length < 2){
                            System.out.println("Usage: consumer <topic> <groupname>");
                            return;
                    }
                    String topic = args[0].toString();
                    String group = args[1].toString();
                    Properties props = new Properties();
                    props.put("bootstrap.servers","10.131.18.138:9092");
                    props.put("group.id",group);
                    props.put("enable.auto.commit","true");
                    props.put("auto.commit.interval.ms","1000");
                    props.put("session.timeout.ms","30000");
                    props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
                    props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserilaizer");
                    KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
                    consumer.subscribe(Arrays.asList(topic));
                    System.out.println("Subscribed to topic");
                    for(String item: Arrays.asList(topic)){
                            System.out.println(item);
                    }
                    while(true){
                            ConsumerRecords<String, String> records = consumer.poll(100);
                            for(ConsumerRecord<String, String> record : records){
                                    System.out.printf("offset = %d, key = %s value = %s\n", record.offset(), record.key(), record.value());
                            }
                    }
            }
    }
    

    相关文章

      网友评论

          本文标题:kafka learning note 2

          本文链接:https://www.haomeiwen.com/subject/orecdhtx.html