0


Kafka

1.Kafka简介

消息队列

  • 消息队列——用于存放消息的组件
  • 程序员可以将消息放入到队列中,也可以从消息队列中获取消息
  • 很多时候消息队列不是一个永久性的存储,是作为临时存储存在的(设定一个期限:设置消息在MQ中保存10天)
  • 消息队列中间件:消息队列的组件,例如:Kafka、Active MQ、RabbitMQ、RocketMQ、ZeroMQ

Kafka的应用场景

  • 异步处理

    • 可以将一些比较耗时的操作放在其他系统中,通过消息队列将需要进行处理的消息进行存储,其他系统可以消费消息队列中的数据- 比较常见的:发送短信验证码、发送邮件

  • 系统解耦

    • 原先一个微服务是通过接口(HTTP)调用另一个微服务,这时候耦合很严重,只要接口发生变化就会导致系统不可用- 使用消息队列可以将系统进行解耦合,现在第一个微服务可以将消息放入到消息队列中,另一个微服务可以从消息队列中把消息取出来进行处理。进行系统解耦

  • 流量削峰

    • 因为消息队列是低延迟、高可靠、高吞吐的,可以应对大量并发

  • 日志处理

    • 可以使用消息队列作为临时存储,或者一种通信管道

消息队列的两种模型

  • 生产者、消费者模型

    • 生产者负责将消息生产到MQ中- 消费者负责从MQ中获取消息- 生产者和消费者是解耦的,可能是生产者一个程序、消费者是另外一个程序
  • 消息队列的模式

    • 点对点:一个消费者消费一个消息- 发布订阅:多个消费者可以消费一个消息

Kafka集群搭建

  • 使用docker安装kafka
version: '3'
services:
  zookeeper-kafka:
    image: wurstmeister/zookeeper
    restart: unless-stopped
    hostname: zookeeper-kafka
    container_name: zookeeper-kafka
    ports:
      - 2181:2181
    volumes:
      - ./data:/data
      - ./datalog:/datalog
  kafka:
    image: wurstmeister/kafka:2.12-2.4.1
    depends_on: 
      - zookeeper-kafka
    container_name: kafka
    ports:
      - 9092:9092
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_BROKER_ID: 0
      KAFKA_ZOOKEEPER_CONNECT: 192.168.200.135:2181      
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.200.135:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
    restart: unless-stopped
    volumes:
      - ./docker.sock:/var/run/docker.sock

ip要改成自己的ip 云服务器需要开放端口 出现报错可以查看docker容器日志排查

  • 搭建集群过程
  1. 将Kafka的安装包上传到虚拟机,并解压

cd /export/software/

tar -xvzf kafka_2.12-2.4.1.tgz -C ../server/

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

  1. 修改 server.properties

cd /export/server/kafka_2.12-2.4.1/config

vim server.properties

指定broker的id

broker.id=0

指定Kafka数据的位置

log.dirs=/export/server/kafka_2.12-2.4.1/data

配置zk的三个节点

zookeeper.connect=node1.itcast.cn:2181,node2.itcast.cn:2181,node3.itcast.cn:2181

  1. 将安装好的kafka复制到另外两台服务器

cd /export/server

scp -r kafka_2.12-2.4.1/ node2.itcast.cn:$PWD

scp -r kafka_2.12-2.4.1/ node3.itcast.cn:$PWD

修改另外两个节点的broker.id分别为1和2

---------node2.itcast.cn--------------

cd /export/server/kafka_2.12-2.4.1/config

vim erver.properties

broker.id=1

--------node3.itcast.cn--------------

cd /export/server/kafka_2.12-2.4.1/config

vim server.properties

broker.id=2

  1. 配置KAFKA_HOME环境变量

vim /etc/profile

export KAFKA_HOME=/export/server/kafka_2.12-2.4.1

export PATH=:$PATH:${KAFKA_HOME}

分发到各个节点

scp /etc/profile node2.itcast.cn:$PWD

scp /etc/profile node3.itcast.cn:$PWD

每个节点加载环境变量

source /etc/profile

  1. 启动服务器

启动ZooKeeper

nohup bin/zookeeper-server-start.sh config/zookeeper.properties &

启动Kafka

cd /export/server/kafka_2.12-2.4.1

nohup bin/kafka-server-start.sh config/server.properties &

测试Kafka集群是否启动成功

bin/kafka-topics.sh --bootstrap-server node1.itcast.cn:9092 --list

  • Kafka集群是必须要有ZooKeeper的

注意:

  • 每一个Kafka的节点都需要修改broker.id(每个节点的标识,不能重复)
  • log.dir数据存储目录需要配置

kafka的目录结构:

目录名称

说明

bin

Kafka的所有执行脚本都在这里。例如:启动Kafka服务器、创建Topic、生产者、消费者程序等等

config

Kafka的所有配置文件

libs

运行Kafka所需要的所有JAR包

logs

Kafka的所有日志文件,如果Kafka出现一些问题,需要到该目录中去查看异常信息

site-docs

Kafka的网站帮助文件

Kafka的生产者/消费者/工具

  • 安装Kafka集群,可以测试以下

    • 创建一个topic主题(消息都是存放在topic中,类似mysql建表的过程)- 基于kafka的内置测试生产者脚本来读取标准输入(键盘输入)的数据,并放入到topic中- 基于kafka的内置测试消费者脚本来消费topic中的数据
  • 推荐大家开发的使用Kafka Tool

    • 浏览Kafka集群节点、多少个topic、多少个分区- 创建topic/删除topic- 浏览ZooKeeper中的数据
  • 工具下载

Offset Explorer

工具使用步骤

Kafka的基准测试工具

  • Kafka中提供了内置的性能测试工具

    • 生产者:测试生产每秒传输的数据量(多少条数据、多少M的数据)5000000 records sent, 11825.446943 records/sec (11.28 MB/sec), 2757.61 ms avg latency- 消费者:测试消费每条拉取的数据量
  • 对比生产者和消费者:消费者的速度更快

  • 参考文档进行测试,根电脑性能有关,一共有三种方式.

Kafka的Eagle监控工具

Kafka Eagle是一款结合了目前大数据Kafka监控工具的特点,重新研发的一块开源免费的Kafka集群优秀的监控工具。它可以非常方便的监控生产环境中的offset、lag变化、partition分布、owner等。

官网: EFAK

采用Docker进行安装

  1. 拉取镜像

docker pull nickzurich/efak:latest

  1. 使用以下命令启动容器(记得开放相关端口)
docker run -d --name kafka-eagle -p 8048:8048 -e EFAK_CLUSTER_ZK_LIST="116.62.237.97:2181" nickzurich/efak:latest

EFAK_CLUSTER_ZK_LIST:配置的是zookeeper服务的地址

  1. 在游览器中输入http://ip:8048 即可访问,初始账号密码: admin/123456

  2. 开启Kafka JMX端口

JMX(Java Management Extensions)是一个为应用程序植入管理功能的框架。JMX是一套标准的代理和服务,实际上,用户可以在任何Java应用程序中使用这些代理和服务实现管理。很多的一些软件都提供了JMX接口,来实现一些管理、监控功能。

在启动Kafka的脚本前,添加:

cd ${KAFKA_HOME}
export JMX_PORT=9988
nohup bin/kafka-server-start.sh config/server.properties &
  1. 安装Kafka-Eagle

安装JDK,并配置好JAVA_HOME。

将kafka_eagle上传,并解压到 /export/server 目录中。

cd cd /export/software/
tar -xvzf kafka-eagle-bin-1.4.6.tar.gz -C ../server/
cd /export/server/kafka-eagle-bin-1.4.6/ 
tar -xvzf kafka-eagle-web-1.4.6-bin.tar.gz
cd /export/server/kafka-eagle-bin-1.4.6/kafka-eagle-web-1.4.6

配置kafka_eagle 环境变量。

vim /etc/profile
export KE_HOME=/export/server/kafka-eagle-bin-1.4.6/kafka-eagle-web-1.4.6
export PATH=$PATH:$KE_HOME/bin
source /etc/profile

配置kafka_eagle。使用vi打开conf目录下的system-config.properties

vim conf/system-config.properties
# 修改第4行,配置kafka集群别名
kafka.eagle.zk.cluster.alias=cluster1
# 修改第5行,配置ZK集群地址
cluster1.zk.list=node1.itcast.cn:2181,node2.itcast.cn:2181,node3.itcast.cn:2181
# 注释第6行
#cluster2.zk.list=xdn10:2181,xdn11:2181,xdn12:2181

# 修改第32行,打开图标统计
kafka.eagle.metrics.charts=true
kafka.eagle.metrics.retain=30

# 注释第69行,取消sqlite数据库连接配置
#kafka.eagle.driver=org.sqlite.JDBC
#kafka.eagle.url=jdbc:sqlite:/hadoop/kafka-eagle/db/ke.db
#kafka.eagle.username=root
#kafka.eagle.password=www.kafka-eagle.org

# 修改第77行,开启mys
kafka.eagle.driver=com.mysql.jdbc.Driver
kafka.eagle.url=jdbc:mysql://node1.itcast.cn:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
kafka.eagle.username=root
kafka.eagle.password=123456

配置JAVA_HOME

cd /export/server/kafka-eagle-bin-1.4.6/kafka-eagle-web-1.4.6/bin
vim ke.sh
# 在第24行添加JAVA_HOME环境配置
export JAVA_HOME=/export/server/jdk1.8.0_241

修改Kafka eagle可执行权限

cd /export/server/kafka-eagle-bin-1.4.6/kafka-eagle-web-1.4.6/bin
chmod +x ke.sh

启动kafka_eagle。

./ke.sh start

访问Kafka eagle,默认用户为admin,密码为:123456

http://ip:8048/ke

Kafka度量指标

topic list

点击Topic下的List菜单,就可以展示当前Kafka集群中的所有topic。

指标

意义

Brokers Spread

broker使用率

Brokers Skew

分区是否倾斜

Brokers Leader Skew

leader partition是否存在倾斜

生产者消息总计

Kafka Java API开发

<!-- kafka客户端工具 -->
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>2.4.1</version>
</dependency>

小tips 将日志输出到文件

  1. 导入log4J的依赖

<**dependency**>
<**groupId**>org.slf4j</**groupId**>
<**artifactId**>slf4j-log4j12</**artifactId**>
<**version**>1.7.6</**version**>
</**dependency**>

<**dependency**>
<**groupId**>log4j</**groupId**>
<**artifactId**>log4j</**artifactId**>
<**version**>1.2.16</**version**>
</**dependency**>

  1. 将log4j.properties配置文件放入到resources文件夹中

log4j.rootLogger=INFO,stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender **
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout **
log4j.appender.stdout.layout.ConversionPattern=
%5p - %m%n

生产者程序开发

kafka 2.4.0 API

  1. 创建连接
    • bootstrap.servers:Kafka的服务器地址- acks:表示当生产者生产数据到Kafka中,Kafka中会以什么样的策略返回- key.serializer:Kafka中的消息是以key、value键值对存储的,而且生产者生产的消息是需要在网络上传到的,这里指定的是StringSerializer方式,就是以字符串方式发送(将来还可以使用其他的一些序列化框架:Google ProtoBuf、Avro)- value.serializer:同上
  1. 创建一个生产者对象KafkaProducer
  2. 调用send方法发送消息(ProducerRecor,封装是key-value键值对)
  3. 调用Future.get表示等带服务端的响应
  4. 关闭生产者
// 1. 创建用于连接Kafka的Properties配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "116.62.237.97:9092");
        props.put("acks", "all");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 2. 创建一个生产者对象KafkaProducer
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);

        // 3. 发送1-100的消息到指定的topic中
        for (int i = 0; i < 100; ++i) {
            // 构建一条消息,直接new ProducerRecord
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", null, i + "");
            Future<RecordMetadata> future = kafkaProducer.send(producerRecord);
            // 调用Future的get方法等待响应
            future.get();
            System.out.println("第" + i + "条消息写入成功!");
        }

        // 4.关闭生产者
        kafkaProducer.close();

消费者程序开发

  • group.id:消费者组的概念,可以在一个消费组中包含多个消费者。如果若干个消费者的group.id是一样的,表示它们就在一个组中,一个组中的消费者是共同消费Kafka中topic的数据。
  • Kafka是一种拉消息模式的消息队列,在消费者中会有一个offset,表示从哪条消息开始拉取数据
  • kafkaConsumer.poll:Kafka的消费者API是一批一批数据的拉取
/**
 * 消费者程序
 * <p>
 * 1.创建Kafka消费者配置
 * Properties props = new Properties();
 * props.setProperty("bootstrap.servers", "node1.itcast.cn: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");
 * <p>
 * 2.创建Kafka消费者
 * 3.订阅要消费的主题
 * 4.使用一个while循环,不断从Kafka的topic中拉取消息
 * 5.将将记录(record)的offset、key、value都打印出来
 */
public class KafkaConsumerOne {
    public static void main(String[] args) {
        // 1.创建Kafka消费者配置
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "116.62.237.97:9092");
        // 消费者组(可以使用消费者组将若干个消费者组织到一起),共同消费Kafka中topic的数据
        // 每一个消费者需要指定一个消费者组,如果消费者的组名是一样的,表示这几个消费者是一个组中的
        props.setProperty("group.id", "test");
        // 自动提交offset
        props.setProperty("enable.auto.commit", "true");
        // 自动提交offset的时间间隔
        props.setProperty("auto.commit.interval.ms", "1000");
        // 拉取的key、value数据的
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 2.创建Kafka消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);

        // 3. 订阅要消费的主题
        // 指定消费者从哪个topic中拉取数据
        kafkaConsumer.subscribe(Arrays.asList("test"));

        // 4.使用一个while循环,不断从Kafka的topic中拉取消息
        while (true) {
            // Kafka的消费者一次拉取一批的数据
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));
            // 5.将将记录(record)的offset、key、value都打印出来
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                // 主题
                String topic = consumerRecord.topic();
                // offset:这条消息处于Kafka分区中的哪个位置
                long offset = consumerRecord.offset();
                // key\value
                String key = consumerRecord.key();
                String value = consumerRecord.value();

                System.out.println("topic: " + topic + " offset:" + offset + " key:" + key + " value:" + value);
            }
        }
    }
}

生产者使用异步方式生产消息

  • 获取生产者消息是否成功,或者成功生产消息到Kafka中后,执行一些其他操作.

  • 使用匿名内部类实现Callback接口,该接口中表示Kafka服务器响应给客户端,会自动调用onCompletion方法

    • metadata:消息的元数据(属于哪个topic、属于哪个partition、对应的offset是什么)- exception:这个对象Kafka生产消息封装了出现的异常,如果为null,表示发送成功,如果不为null,表示出现异常。
// 二、使用异步回调的方式发送消息
ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", null, i + "");
kafkaProducer.send(producerRecord, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        // 1. 判断发送消息是否成功
        if(exception == null) {
            // 发送成功
            // 主题
            String topic = metadata.topic();
            // 分区id
            int partition = metadata.partition();
            // 偏移量
            long offset = metadata.offset();
            System.out.println("topic:" + topic + " 分区id:" + partition + " 偏移量:" + offset);
        }
        else {
            // 发送出现错误
            System.out.println("生产消息出现异常!");
            // 打印异常消息
            System.out.println(exception.getMessage());
            // 打印调用栈
            System.out.println(exception.getStackTrace());
        }
    }
});

Kafka中的重要概念

  • broker

    • Kafka服务器进程,生产者、消费者都要连接broker- 一个集群由多个broker组成,功能实现Kafka集群的负载均衡、容错

  • producer:生产者

  • consumer:消费者

  • topic:主题,一个Kafka集群中,可以包含多个topic。一个topic可以包含多个分区

    • 是一个逻辑结构,生产、消费消息都需要指定topic
  • partition:Kafka集群的分布式就是由分区来实现的。一个topic中的消息可以分布在topic中的不同partition中

  • replica:副本,实现Kafkaf集群的容错,实现partition的容错。一个topic至少应该包含大于1个的副本

  • consumer group:消费者组,一个消费者组中的消费者可以共同消费topic中的分区数据。每一个消费者组都一个唯一的名字。配置group.id一样的消费者是属于同一个组中

  • offset:偏移量。相对消费者、partition来说,可以通过offset来拉取数据

消费者组

  • 一个消费者组中可以包含多个消费者,共同来消费topic中的数据
  • 一个topic中如果只有一个分区,那么这个分区只能被某个组中的一个消费者消费
  • 有多少个分区,那么就可以被同一个组内的多少个消费者消费

幂等性

  • 生产者消息重复问题

    • Kafka生产者生产消息到partition,如果直接发送消息,kafka会将消息保存到分区中,但Kafka会返回一个ack给生产者,表示当前操作是否成功,是否已经保存了这条消息。如果ack响应的过程失败了,此时生产者会重试,继续发送没有发送成功的消息,Kafka又会保存一条一模一样的消息

  • 在Kafka中可以开启幂等性
props.put("enable.idempotence",true);
    • 当Kafka的生产者生产消息时,会增加一个pid(生产者的唯一编号)和sequence number(针对消息的一个递增序列)- 发送消息,会连着pid和sequence number一块发送- kafka接收到消息,会将消息和pid、sequence number一并保存下来- 如果ack响应失败,生产者重试,再次发送消息时,Kafka会根据pid、sequence number是否需要再保存一条消息- 判断条件:生产者发送过来的sequence number 是否小于等于 partition中消息对应的sequence

事务编程

  • Kafka事务指的是生产者生产消息以及消费者提交offset的操作可以在一个原子操作中,要么都成功,要么都失败。

  • 开启事务的条件

    • 生产者
// 开启事务必须要配置事务的ID
props.put("transactional.id", "dwd_user");
    • 消费者
// 配置事务的隔离级别
props.put("isolation.level","read_committed");
// 关闭自动提交,一会我们需要手动来提交offset,通过事务来维护offset
props.setProperty("enable.auto.commit", "false");
    • 生产者
      • 初始化事务 initTransactions- 开启事务 beginTransaction- 需要使用producer来将消费者的offset提交到事务中 sendOffsetsToTransaction- 提交事务 commitTransaction- 如果出现异常回滚事务 abortTransaction

如果使用了事务,不要使用异步发送

public class TransactionProgram {
    public static void main(String[] args) {
        // 1. 调用之前实现的方法,创建消费者、生产者对象
        KafkaConsumer<String, String> consumer = createConsumer();
        KafkaProducer<String, String> producer = createProducer();

        // 2. 生产者调用initTransactions初始化事务
        producer.initTransactions();

        // 3. 编写一个while死循环,在while循环中不断拉取数据,进行处理后,再写入到指定的topic
        while(true) {
            try {
                // (1)    生产者开启事务
                producer.beginTransaction();

                // 这个Map保存了topic对应的partition的偏移量
                Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();

                // 从topic中拉取一批的数据
                // (2)    消费者拉取消息
                ConsumerRecords<String, String> concumserRecordArray = consumer.poll(Duration.ofSeconds(5));
                // (3)    遍历拉取到的消息,并进行预处理
                for (ConsumerRecord<String, String> cr : concumserRecordArray) {
                    // 将1转换为男,0转换为女
                    String msg = cr.value();
                    String[] fieldArray = msg.split(",");

                    // 将消息的偏移量保存
                    // 消费的是ods_user中的数据
                    String topic = cr.topic();
                    int partition = cr.partition();
                    long offset = cr.offset();

                    int i = 1 / 0;

                    // offset + 1:offset是当前消费的记录(消息)对应在partition中的offset,而我们希望下一次能继续从下一个消息消息
                    // 必须要+1,从能消费下一条消息
                    offsetMap.put(new TopicPartition(topic, partition), new OffsetAndMetadata(offset + 1));

                    // 将字段进行替换
                    if(fieldArray != null && fieldArray.length > 2) {
                        String sexField = fieldArray[1];
                        if(sexField.equals("1")) {
                            fieldArray[1] = "男";
                        }
                        else if(sexField.equals("0")){
                            fieldArray[1] = "女";
                        }
                    }

                    // 重新拼接字段
                    msg = fieldArray[0] + "," + fieldArray[1] + "," + fieldArray[2];

                    // (4)    生产消息到dwd_user topic中
                    ProducerRecord<String, String> dwdMsg = new ProducerRecord<>("dwd_user", msg);
                    // 发送消息
                    Future<RecordMetadata> future = producer.send(dwdMsg);
                    try {
                        future.get();
                    } catch (Exception e) {
                        e.printStackTrace();
                        producer.abortTransaction();
                    }
//                            new Callback()
//                    {
//                        @Override
//                        public void onCompletion(RecordMetadata metadata, Exception exception) {
//                            // 生产消息没有问题
//                            if(exception == null) {
//                                System.out.println("发送成功:" + dwdMsg);
//                            }
//                            else {
//                                System.out.println("生产消息失败:");
//                                System.out.println(exception.getMessage());
//                                System.out.println(exception.getStackTrace());
//                            }
//                        }
//                    });
                }

                producer.sendOffsetsToTransaction(offsetMap, "ods_user");

                // (6)    提交事务
                producer.commitTransaction();
            }catch (Exception e) {
                e.printStackTrace();
                // (7)    捕获异常,如果出现异常,则取消事务
                producer.abortTransaction();
            }
        }
    }

    // 一、创建一个消费者来消费ods_user中的数据
    private static KafkaConsumer<String, String> createConsumer() {
        // 1. 配置消费者的属性(添加对事务的支持)
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "node1.itcast.cn:9092");
        props.setProperty("group.id", "ods_user");
        // 配置事务的隔离级别
        props.put("isolation.level","read_committed");
        // 关闭自动提交,一会我们需要手动来提交offset,通过事务来维护offset
        props.setProperty("enable.auto.commit", "false");
        // 反序列化器
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 2. 构建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);

        // 3. 订阅一个topic
        kafkaConsumer.subscribe(Arrays.asList("ods_user"));

        return kafkaConsumer;

    }

    // 二、编写createProducer方法,用来创建一个带有事务配置的生产者
    private static KafkaProducer<String, String> createProducer() {
        // 1. 配置生产者带有事务配置的属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "node1.itcast.cn:9092");
        // 开启事务必须要配置事务的ID
        props.put("transactional.id", "dwd_user");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 2. 构建生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);

        return kafkaProducer;
    }
}

Kafka中的分区副本机制

生产者的分区写入策略

  • 轮询(按照消息尽量保证每个分区的负载)策略,消息会均匀地分布到每个partition

    • 写入消息的时候,key为null的时候,默认使用的是轮询策略- 默认的策略,也是使用最多的策略,可以最大限度保证所有消息平均分配到一个分区

  • 随机策略(不使用)
  • 按key写入策略,key.hash() % 分区的数量

  • 自定义分区策略(类似于MapReduce指定分区)

乱序问题

  • 在Kafka中生产者是有写入策略,如果topic有多个分区,就会将数据分散在不同的partition中存储
  • 当partition数量大于1的时候,数据(消息)会打散分布在不同的partition中
  • 如果只有一个分区,消息是有序的

自定义分区策略:

  1. 创建自定义分区器
public class KeyWithRandomPartitioner implements Partitioner {

    private Random r;

    @Override
    public void configure(Map<String, ?> configs) {
        r = new Random();
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // cluster.partitionCountForTopic 表示获取指定topic的分区数量
        return r.nextInt(1000) % cluster.partitionCountForTopic(topic);
    }

    @Override
    public void close() {
    }
}
  1. 在Kafka生产者配置中,自定使用自定义分区器的类名
props.
put(ProducerConfig.PARTITIONER_CLASS_CONFIG, KeyWithRandomPartitioner.class.getName());

消费组Consumer Group Rebalance机制

  • 再均衡:在某些情况下,消费者组中的消费者消费的分区会产生变化,会导致消费者分配不均匀(例如:有两个消费者消费3个,因为某个partition崩溃了,还有一个消费者当前没有分区要削峰),Kafka Consumer Group就会启用rebalance机制,重新平衡这个Consumer Group内的消费者消费的分区分配。

  • 触发时机

    • 消费者数量发生变化
      • 某个消费者crash- 新增消费者
    • topic的数量发生变化
      • 某个topic被删除
    • partition的数量发生变化
      • 删除partition- 新增partition
  • 不良影响

    • 发生rebalance,所有的consumer将不再工作,共同来参与再均衡,直到每个消费者都已经被成功分配所需要消费的分区为止(rebalance结束)

消费者的分区分配策略

分区分配策略:保障每个消费者尽量能够均衡地消费分区的数据,不能出现某个消费者消费分区的数量特别多,某个消费者消费的分区特别少

  • Range分配策略(范围分配策略):Kafka默认的分配策略

    • n:分区的数量 / 消费者数量- m:分区的数量 % 消费者数量- 前m个消费者消费n+1个分区- 剩余的消费者消费n个分区

  • RoundRobin分配策略(轮询分配策略)

    • 消费者挨个分配消费的分区- 配置消费者的partition.assignment.strategy为org.apache.kafka.clients.consumer.RoundRobinAssignor。

  • Striky粘性分配策略

    • 在没有发生rebalance跟轮询分配策略是一致的- 发生了rebalance,轮询分配策略,重新走一遍轮询分配的过程。而粘性会保证跟上一次的尽量一致,只是将新的需要分配的分区,均匀的分配到现有可用的消费者中即可- 减少上下文的切换- Striky粘性分配策略,保留rebalance之前的分配结果。

副本的ACK机制

producer是不断地往Kafka中写入数据,写入数据会有一个返回结果,表示是否写入成功。这里对应有一个ACKs的配置。

Properties props = new Properties();
props.put("bootstrap.servers", "node1.itcast.cn:9092");
props.put("acks", "all");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  • acks = 0:生产者只管写入,不管是否写入成功,可能会数据丢失。性能是最好的

  • acks = 1:生产者会等到leader分区写入成功后,返回成功,接着发送下一条

  • acks = -1/all:确保消息写入到leader分区、还确保消息写入到对应副本都成功后,接着发送下一条,性能是最差的

根据业务情况来选择ack机制,是要求性能最高,一部分数据丢失影响不大,可以选择0/1。如果要求数据一定不能丢失,就得配置为-1/all。

分区中是有leader和follower的概念,为了确保消费者消费的数据是一致的,只能从分区leader去读写消息,follower做的事情就是同步数据,Backup。

高级API(High-Level API)、低级API(Low-Level API)

  • 高级API就是直接让Kafka帮助管理、处理分配、数据

    • offset存储在ZK中- 由kafka的rebalance来控制消费者分配的分区- 开发起来比较简单,无需开发者关注底层细节- 无法做到细粒度的控制
/**
 * 消费者程序:从test主题中消费数据
 */
public class _2ConsumerTest {
    public static void main(String[] args) {
        // 1. 创建Kafka消费者配置
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "192.168.88.100: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");

        // 2. 创建Kafka消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 3. 订阅要消费的主题
        consumer.subscribe(Arrays.asList("test"));

        // 4. 使用一个while循环,不断从Kafka的topic中拉取消息
        while (true) {
            // 定义100毫秒超时
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
    }
}
  • 低级API:由编写的程序自己控制逻辑

    • 自己来管理Offset,可以将offset存储在ZK、MySQL、Redis、HBase、Flink的状态存储- 指定消费者拉取某个分区的数据- 可以做到细粒度的控制- 原有的Kafka的策略会失效,需要我们自己来实现消费机制
String topic = "test";
TopicPartition partition0 = new TopicPartition(topic, 0);
TopicPartition partition1 = new TopicPartition(topic, 1);
consumer.assign(Arrays.asList(partition0, partition1));

Kafka原理

leader和follower

  • Kafka中的leader和follower是相对分区有意义,不是相对broker

  • Kafka在创建topic的时候,会尽量分配分区的leader在不同的broker中,其实就是负载均衡

  • leader职责:读写数据

  • follower职责:同步数据、参与选举(leader crash之后,会选举一个follower重新成为分区的leader

  • 注意和ZooKeeper区分

    • ZK的leader负责读、写,follower可以读取- Kafka的leader负责读写、follower不能读写数据(确保每个消费者消费的数据是一致的),Kafka一个topic有多个分区leader,一样可以实现数据操作的负载均衡.

AR\ISR\OSR

  • AR表示一个topic下的所有副本
  • ISR:In Sync Replicas,正在同步的副本(可以理解为当前有几个follower是存活的)
  • OSR:Out of Sync Replicas,不再同步的副本
  • AR = ISR + OSR

leader选举

  • Controller:controller是kafka集群的老大,是针对Broker的一个角色

    • Controller是高可用的,是用过ZK来进行选举
  • Leader:是针对partition的一个角色

    • Leader是通过ISR来进行快速选举
  • 如果Kafka是基于ZK来进行选举,ZK的压力可能会比较大。例如:某个节点崩溃,这个节点上不仅仅只有一个leader,是有不少的leader需要选举。通过ISR快速进行选举。

  • leader的负载均衡

    • 如果某个broker crash之后,就可能会导致partition的leader分布不均匀,就是一个broker上存在一个topic下不同partition的leader- 通过以下指令,可以将leader分配到优先的leader对应的broker,确保leader是均匀分配的
bin/kafka-leader-election.sh --bootstrap-server node1.itcast.cn:9092 --topic test --partition=2 --election-type preferred

Kafka读写流程

  • 写流程

    • 通过ZooKeeper找partition对应的leader,leader是负责写的- producer开始写入数据- ISR里面的follower开始同步数据,并返回给leader ACK- 返回给producer ACK
  • 读流程

    • 通过ZooKeeper找partition对应的leader,leader是负责读的- 通过ZooKeeper找到消费者对应的offset- 然后开始从offset往后顺序拉取数据- 提交offset(自动提交——每隔多少秒提交一次offset、手动提交——放入到事务中提交)

Kafka的物理存储

  • Kafka的数据组织结构

    • topic- partition- segment
      • .log数据文件- .index(稀疏索引)- .timeindex(根据时间做的索引)
  • 深入了解读数据的流程

    • 消费者的offset是一个针对partition全局offset- 可以根据这个offset找到segment段- 接着需要将全局的offset转换成segment的局部offset- 根据局部的offset,就可以从(.index稀疏索引)找到对应的数据位置- 开始顺序读取

消息传递的语义性

Flink里面有对应的每种不同机制的保证,提供Exactly-Once保障(二阶段事务提交方式)

  • At-most once:最多一次(只管把数据消费到,不管有没有成功,可能会有数据丢失)
  • At-least once:最少一次(有可能会出现重复消费)
  • Exactly-Once:仅有一次(事务性性的保障,保证消息有且仅被处理一次)

Kafka的消息不丢失

  • broker消息不丢失:因为有副本relicas的存在,会不断地从leader中同步副本,所以,一个broker crash,不会导致数据丢失,除非是只有一个副本。

  • 生产者消息不丢失:ACK机制(配置成ALL/-1)、配置0或者1有可能会存在丢失

  • 消费者消费不丢失:重点控制offset

    • At-least once:一种数据可能会重复消费- Exactly-Once:仅被一次消费

数据积压

  • 数据积压指的是消费者因为有一些外部的IO、一些比较耗时的操作(Full GC——Stop the world),就会造成消息在partition中一直存在得不到消费,就会产生数据积压
  • 在企业中,我们要有监控系统,如果出现这种情况,需要尽快处理。虽然后续的Spark Streaming/Flink可以实现背压机制,但是数据累积太多一定对实时系统它的实时性是有说影响的

数据清理&配额限速

  • 数据清理

    • Log Deletion(日志删除):如果消息达到一定的条件(时间、日志大小、offset大小),Kafka就会自动将日志设置为待删除(segment端的后缀名会以 .delete结尾),日志管理程序会定期清理这些日志
      • 默认是7天过期
    • Log Compaction(日志合并)
      • 如果在一些key-value数据中,一个key可以对应多个不同版本的value- 经过日志合并,就会只保留最新的一个版本
  • 配额限速

    • 可以限制Producer、Consumer的速率- 防止Kafka的速度过快,占用整个服务器(broker)的所有IO资源
标签: kafka java-ee

本文转载自: https://blog.csdn.net/m0_51607909/article/details/136760806
版权归原作者 网络深处的易某某 所有, 如有侵权,请联系我们删除。

“Kafka”的评论:

还没有评论