day04kafka_大数据系统

一. 复习

kafka使用分布式公平架构，主节点：kafka controllere （负责存储和管理）从节点：kafka broker（负责存储）如果主节点挂掉，会依赖zk重新选举。
kafka的数据安全是依赖副本机制
leader和follwer是topic下的part的主节点和从节点，而controller和broker是集群的

二.topic管理：创建与列举、

/export/server/kafka_2.12-2.4.1/bin

三.生产者及消费者测试

四.kafka集群压力测试

创建topic：bigdata ：kafka-topics.sh --create --topic bigdata --partitions 2 --replication-factor 2 --bootstrap-server node1:9092,node2:9092,node3:9092

生产测试： kafka-producer-perf-test.sh --topic bigdata --num-records 100000 --throughput -1 --record-size 1000 --producer-props bootstrap.servers=node1:9092,node2:9092,node3:9092 acks=1

消费测试： kafka-consumer-perf-test.sh --topic bigdata --broker-list node1:9092,node2:9092,node3:9092 --fetch-size 1048576 --messages 100000

五.kafka API



aliyun
http://maven.aliyun.com/nexus/content/groups/public/






org.apache.kafka
kafka-clients
2.4.1

2.2 代码

package bigdata.hlzq.com.kafka.produce;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

//生产者数据到kafka
public class KafkaProducerTestClient {
    public static void main(String[] args) {
        //构建连接
        //构建一个配置对象
        Properties props = new Properties();
        props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");//服务端地址
        props.put("acks","all");//生产者写入网络部丢失的原因ack+重试机制：写入一条到kafka分区，kafka会返回一个ack确认，如果没有返回重新发送ack的值： 0不用等待 1：等待写入到lead就返回 all：等待所有副本
        //定义写入kafka的类型
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //构建连接对象加载配置
        KafkaProducer producer = new KafkaProducer<>(props);
        //实现操作
        for (int i=1;i<=100;i++){
            //调用连接对象的方法，指定topic key决定分区规则
            producer.send(new ProducerRecord("bigdata01",i+"",i+"itc"));
            //不给key在同一个分区
            producer.send(new ProducerRecord("bigdata01",i+"hh"));
            //指定分区
            producer.send(new ProducerRecord("bigdata01", 1,i+"",i+"itc"));
        }
        //释放连接
        producer.close();
    }
}

import java.util.Properties;
import java.util.function.Consumer;

public class KafkaCons {
    public static void main(String[] args) {
        // 构建消费者连接
        //构建配置对象
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        props.setProperty("group.id", "test");
        props.setProperty("enable.auto.commit", "true");
        props.setProperty("auto.commit.interval.ms","1000");
        props.setProperty("key.deserializer" , "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer consumer = new KafkaConsumer(props);
        // 订阅消费处理
        consumer.subscribe(Arrays.asList("bigdata01"));
        //消费者是不停的
        while (true){
            ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecordrecord:records){
                System.out.println("offset:"+record.offset()+"   "+"topic:"+record.topic()+"  "+"part:"+record.partition()+"  "+"value:"+record.value());

            }
        }
    }
}

六. 生产分区规则

数据存储在那个分区的规则，

 //调用连接对象的方法，指定topic key决定分区规则
            producer.send(new ProducerRecord("bigdata01",i+"",i+"itc"));
            //不给key在同一个分区
            producer.send(new ProducerRecord("bigdata01",i+"hh"));
            //指定分区
            producer.send(new ProducerRecord("bigdata01", 1,i+"",i+"itc"));

默认分区器defaultpartitioner
如果指定了key：按照key的hash取余分区的个数，来写入对应的分区：只要key一样就会进入同一个分区，容易导致数据倾斜
黏性分区器：实现少批次多数据：一个批次的数据都存放在一个分区。（判断缓存中是否有这个topic的分区连接，如果有直接使用，没有随机写入一个分区，并放入缓存）
轮循分区：数据分配相对均衡，批次多，每个批次数据量少，性能差。（不用）

分区规则：先判断有没有指定分区，指定分区就写入指定的分区，调用分区器在判断有没有指定key如果有key按照key类似于hash取余的方式计算分区，没有的话使用黏性分区自定义开发生产分区器

package bigdata.hlzq.com.kafka.userpart;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;
import java.util.Random;

//自定义分区器
public class UserPartition implements Partitioner {
  
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
       //获取topic的分区个数
        Integer integer = cluster.partitionCountForTopic(topic);
        //构建随机分区随机值
        Random random = new Random();
        int i = random.nextInt(integer);
        return i;
    }

    @Override
    public void close() {
    //释放资源连接
    }

    @Override
    public void configure(Map configs) {
     //获取配置
    }
}
在生产者代码中指定分区配置
 props.put("partitioner.class", "bigdata.hlzq.com.kafka.userpart.UserPartition");

七.消费者消费安全问题

day04kafka

大数据系统相关栏目本月热门文章