kafka的集群操作和API的操作

一、集群操作

1、创建topic

   创建一个名字为test的主题， 有三个分区，有两个副本

    hadoop01执行以下命令来执行

帮助命令 
进入bin/kafka-topics.sh   help
cd /export/servers/kafka_2.11-1.0.0
bin/kafka-topics.sh --create --zookeeper hadoop01:2181 --replication-factor 2 --partitions 3 --topic test

2、查看主题命令

  查看kafka当中存在的主题

  hadoop01使用以下命令来查看kafka当中存在的topic主题

cd /export/servers/kafka_2.11-1.0.0
bin/kafka-topics.sh  --list --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181

3、生产者生产数据

  模拟生产者来生产数据

  hadoop01服务器执行以下命令模拟生产者进行生产数据

cd /export/servers/kafka_2.11-1.0.0
bin/kafka-console-producer.sh --broker-list hadoop01:9092,hadoop02:9092,hadoop03:9092 --topic test

4、消费者消费数据

  hadoop02服务器或者开个hadoop01执行

cd /export/servers/kafka_2.11-1.0.0
bin/ kafka-console-consumer.sh --from-beginning --topic test  --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181

5、运行describe topics命令

  hadoop01执行以下命令运行describe查看topic的相关信息

cd /export/servers/kafka_2.11-1.0.0
bin/kafka-topics.sh --describe --zookeeper hadoop01:2181 --topic test



结果说明：
这是输出的解释。第一行给出了所有分区的摘要，每个附加行提供有关一个分区的信息。由于我们只有一个分 区用于此主题，因此只有一行。

“leader”是负责给定分区的所有读取和写入的节点。每个节点将成为随机选择的分区部分的领导者。（因为在kafka中 如果有多个副本的话，就会存在leader和follower的关系，表示当前这个副本为leader所在的broker是哪一个） 
“replicas”是复制此分区日志的节点列表，无论它们是否为领导者，或者即使它们当前处于活动状态。（所有副本列表0,1,2） 
“isr”是“同步”复制品的集合。这是副本列表的子集，该列表当前处于活跃状态并且已经被领导者捕获。（可用的列表数）

6、增加topic分区数

  任意kafka服务器执行以下命令可以增加topic分区数

cd /export/servers/kafka_2.11-1.0.0
bin/kafka-topics.sh --zookeeper zkhost:port --alter --topic topicName --partitions 8

7、增加配置

  动态修改kakfa的配置

cd /export/servers/kafka_2.11-1.0.0
bin/kafka-topics.sh --zookeeper hadoop01:2181 --alter --topic test --config flush.messages=1

8、删除配置

  动态删除kafka集群配置

cd /export/servers/kafka_2.11-1.0.0
bin/kafka-topics.sh --zookeeper hadoop01:2181 --alter --topic test --delete-config flush.messages

9、删除topic

  目前删除topic在默认情况下知识打上一个删除的标记，在重新启动kafka后才删除。如果需要立即删除，则需要在

  server.properties中配置：

  delete.topic.enable=true

  然后执行以下命令进行删除topic

kafka-topics.sh --zookeeper zkhost:port --delete --topic topicName

二、API操作

  1、使用生产者生产数据

public class MyProducer {
    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092");
        // 针对保存数据的确认
        props.put("acks", "all");
        // 尝试重试的次数
        props.put("retries", 0);
        // 按照batch发送数据
        props.put("batch.size", 16384);
        // 消息发送的延时时间
        props.put("linger.ms", 1);
        // 消息异步发送的存储消息的缓存
        props.put("buffer.memory", 33554432);
        // 指定key和value的序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        for (int i = 0; i <100 ; i++) {
            // 构造一条消息队列
            // 第一种分区策略，如果既没有指定分区号，也没有指定数据key，那么就会使用轮询的方式将数据均匀的发送到不同的分区里面去
            ProducerRecord<String, String> record = new ProducerRecord<String, String>("test", "hello world" + i);
            //2.按照key的hash分区
            // 如果没有指定分区号，指定了数据key，通过key.hashCode  % numPartitions来计算数据究竟会保存在哪一个分区里面
            //ProducerRecord<String, String> record = new ProducerRecord<String, String>("test",""+i,"hello world"+i);
            //3.指定分区
            //如果指定了分区号，那么就会将数据直接写入到对应的分区里面去
            //ProducerRecord<String, String> record = new ProducerRecord<String, String>("test",1,""+i,"hello world"+i);
            //4.自定义分区
            //将数据发送给kafka
            //自定义分区策略。如果不自定义分区规则，那么会将数据使用轮询的方式均匀的发送到各个分区里面去
            //ProducerRecord<String, String> record = new ProducerRecord<String, String>("test","hello world"+i);
            // 通过send发送将数据保存起来
            producer.send(record);
        }
        producer.close();
    }

2、消费者消费数据

public class MyConsumer {

    public static void main(String[] args) {

        Properties props = new Properties();
        //broker节点
        props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092");
        //消费者组
        props.put("group.id", "itcast");
        //是否自动提交offset偏移量
        props.put("enable.auto.commit", "true");
        //500ms
        //每次地洞提交偏移量的时间
        props.put("auto.commit.interval.ms", "1000");
        //将数据进行反序列化
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        //订阅主题
        consumer.subscribe(Arrays.asList("test"));

        while (true){
            //获取到消费的数据集合
            ConsumerRecords<String, String> records = consumer.poll(3000);
            for (ConsumerRecord<String, String> record : records) {
                //获取每一条记录
                String topic = record.topic();
                long offset = record.offset();
                String value = record.value();
                int partition = record.partition();
                System.out.println("topic:"+topic+",partition:"+partition+",offset:"+offset+",value:"+value);
            }

        }

    }
}

3、指定分区进行消费

   //订阅主题和指定分区消费，二者只能出现其一
        TopicPartition partition0 = new TopicPartition("test", 0);
        TopicPartition partition1 = new TopicPartition("test", 1);
        consumer.assign(Arrays.asList(partition0,partition1));