时间：   |   分类：     | 字数： 6997 字 | 阅读约： 14分钟 | 阅读次数：

Kafka架构与实战

1. 概念和基本架构

1.1 介绍

Kafka是最初由Linkedin公司开发，是⼀个分布式、分区的、多副本的、多⽣产者、多订阅者，基于zookeeper协调的分布式⽇志系统（也可以当做MQ系统），常⻅可以⽤于web/nginx⽇志、访问⽇志，消息服务等等，Linkedin于2010年贡献给了Apache基⾦会并成为顶级开源项⽬。

应用场景: 日志收集系统, 消息系统

主要设计目标如下:

  • 以时间复杂度为O(1)的⽅式提供消息持久化能⼒，即使对TB级以上数据也能保证常数时间的访问性能。
  • ⾼吞吐率。即使在⾮常廉价的商⽤机器上也能做到单机⽀持每秒100K条消息的传输。
  • ⽀持KafkaServer间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
  • 同时⽀持离线数据处理和实时数据处理。
  • ⽀持在线⽔平扩展

  1. Kafka在⼀个或多个可以跨越多个数据中⼼的服务器上作为集群运⾏。
  2. Kafka集群中按照主题分类管理，⼀个主题可以有多个分区，⼀个分区可以有多个副本分区。
  3. 每个记录由⼀个键，⼀个值和⼀个时间戳组成。

有两种主要消息传递模式:

  • 点对点传递模式

  • 发布-订阅模式

    大部分系统主要采用这种模式

Kafka只有消息的拉取, 没有推送,可以通过轮询实现消息的推送

4个核心API:

  • Producer API: 允许应⽤程序将记录流发布到⼀个或多个Kafka主题
  • Consumer API: 允许应⽤程序订阅⼀个或多个主题并处理为其⽣成的记录流
  • Streams API：允许应⽤程序充当流处理器，使⽤⼀个或多个主题的输⼊流，并⽣成⼀个或多个输出主题的输出流，从⽽有效地将输⼊流转换为输出流。
  • Connector API：允许构建和运⾏将Kafka主题连接到现有应⽤程序或数据系统的可重⽤⽣产者或使⽤者。例如，关系数据库的连接器可能会捕获对表的所有更改。

1.2 优势

  • ⾼吞吐量：单机每秒处理⼏⼗上百万的消息量。即使存储了许多TB的消息，它也保持稳定的性能。
  • ⾼性能：单节点⽀持上千个客户端，并保证零停机和零数据丢失。
  • 持久化数据存储：将消息持久化到磁盘。通过将数据持久化到硬盘以及replication防⽌数据丢失。
    • 零拷贝
    • 顺序读, 顺序写
    • 利用linux的页缓存
  • 分布式系统，易于向外扩展。所有的Producer、Broker和Consumer都会有多个，均为分布式的。⽆需停机即可扩展机器。多个Producer、Consumer可能是不同的应⽤。
  • 可靠性-Kafka是分布式，分区，复制和容错的。
  • 客户端状态维护：消息被处理的状态是在Consumer端维护，⽽不是由server端维护。当失败时能⾃动平衡
  • ⽀持online和offline的场景。
  • ⽀持多种客户端语⾔。Kafka⽀持Java、.NET、PHP、Python等多种语⾔。

1.3 应用场景

  • 日志收集

  • 消息系统

  • 用户活动跟踪

    Kafka经常被⽤来记录Web⽤户或者App⽤户的各种活动，如浏览⽹⻚、搜索、点击等活动，这些活动信息被各个服务器发布到Kafka的Topic中，然后消费者通过订阅这些Topic来做实时的监控分析，亦可保存到数据库；

  • 运营指标

    Kafka也经常⽤来记录运营监控数据。包括收集各种分布式应⽤的数据，⽣产各种操作的集中反馈，⽐如报警和报告；

  • 流式处理

    ⽐如SparkStreaming和Storm。

1.4 基本架构

  • 消息和批次

    Kafka的数据单元称为消息。可以把消息看成是数据库⾥的⼀个“数据⾏”或⼀条“记录”。消息由字节数组组成。

    消息有键，键也是⼀个字节数组。当消息以⼀种可控的⽅式写⼊不同的分区时，会⽤到键。

    为了提⾼效率，消息被分批写⼊Kafka。批次就是⼀组消息，这些消息属于同⼀个主题和分区。把消息分成批次可以减少⽹络开销。批次越⼤，单位时间内处理的消息就越多，单个消息的传输时间就越⻓。批次数据会被压缩，这样可以提升数据的传输和存储能⼒，但是需要更多的计算处理。

  • 模式

    消息模式（schema）有许多可⽤的选项，以便于理解。如JSON和XML，但是它们缺乏强类型处理能⼒。Kafka的许多开发者喜欢使⽤Apache Avro。Avro提供了⼀种紧凑的序列化格式，模式和消息体分开。当模式发⽣变化时，不需要重新⽣成代码，它还⽀持强类型和模式进化，其版本既向前兼容，也向后兼容。

    数据格式的⼀致性对Kafka很重要，因为它消除了消息读写操作之间的耦合性

  • 主题和分区

    Kafka的消息通过主题进⾏分类。主题可⽐是数据库的表或者⽂件系统⾥的⽂件夹。主题可以被分为若⼲分区，⼀个主题通过分区分布于Kafka集群中，提供了横向扩展的能⼒。

  • 生产者和消费者

    ⽣产者创建消息。消费者消费消息。⼀个消息被发布到⼀个特定的主题上。⽣产者在默认情况下把消息均衡地分布到主题的所有分区上：

    • 直接指定消息的分区

    • 根据消息的key散列取模得出分区

    • 轮询指定分区。

    消费者通过偏移量来区分已经读过的消息，从⽽消费消息。消费者是消费组的⼀部分。消费组保证每个分区只能被⼀个消费者使⽤，避免重复消费。

  • Broker和集群

    ⼀个独⽴的Kafka服务器称为broker。broker接收来⾃⽣产者的消息，为消息设置偏移量，并提交消息到磁盘保存。broker为消费者提供服务，对读取分区的请求做出响应，返回已经提交到磁盘上的消息。单个broker可以轻松处理数千个分区以及每秒百万级的消息量。

    每个集群都有⼀个broker是集群控制器（⾃动从集群的活跃成员中选举出来）。控制器负责管理⼯作：

    • 将分区分配给broker
    • 监控broker

    集群中⼀个分区属于⼀个broker，该broker称为分区⾸领。⼀个分区可以分配给多个broker，此时会发⽣分区复制。分区的复制提供了消息冗余，⾼可⽤。副本分区不负责处理消息的读写。

1.5 核心概念

  • Producer

    ⽣产者创建消息。该⻆⾊将消息发布到Kafka的topic中。broker接收到⽣产者发送的消息后，broker将该消息追加到当前⽤于追加数据的segment文件中

    ⼀般情况下，⼀个消息会被发布到⼀个特定的主题上。

    • 默认情况下通过轮询把消息均衡地分布到主题的所有分区上。
    • 在某些情况下，⽣产者会把消息直接写到指定的分区。这通常是通过消息键和分区器来实现的，分区器为键⽣成⼀个散列值，并将其映射到指定的分区上。这样可以保证包含同⼀个键的消息会被写到同⼀个分区上。
    • ⽣产者也可以使⽤⾃定义的分区器，根据不同的业务规则将消息映射到分区

  • Consumer

    消费者读取消息。

    • 消费者订阅⼀个或多个主题，并按照消息⽣成的顺序读取它们。
    • 消费者通过检查消息的偏移量来区分已经读取过的消息。偏移量是另⼀种元数据，它是⼀个不断递增的整数值，在创建消息时，Kafka会把它添加到消息⾥。在给定的分区⾥，每个消息的偏移量都是唯⼀的。消费者把每个分区最后读取的消息偏移量保存在Zookeeper或Kafka上，如果消费者关闭或重启，它的读取状态不会丢失。
    • 消费者是消费组的⼀部分。群组保证每个分区只能被⼀个消费者使⽤。
    • 如果⼀个消费者失效，消费组⾥的其他消费者可以接管失效消费者的⼯作，再平衡，分区重新分配。

  • Broker

    ⼀个独⽴的Kafka服务器被称为broker。broker为消费者提供服务，对读取分区的请求作出响应，返回已经提交到磁盘上的消息。

    • 如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的⼀个partition。
    • 如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的⼀个partition，剩下的M个broker不存储该topic的partition数据。
    • 如果某topic有N个partition，集群中broker数⽬少于N个，那么⼀个broker存储该topic的⼀个或多个partition。在实际⽣产环境中，尽量避免这种情况的发⽣，这种情况容易导致Kafka集群数据不均衡。

    broker是集群的组成部分。每个集群都有⼀个broker同时充当了集群控制器的⻆⾊（⾃动从集群的活跃成员中选举出来）。

    控制器负责管理⼯作，包括将分区分配给broker和监控broker。

    在集群中，⼀个分区从属于⼀个broker，该broker被称为分区的⾸领。

  • Topic

    每条发布到Kafka集群的消息都有⼀个类别，这个类别被称为Topic。物理上不同Topic的消息分开存储。主题就好⽐数据库的表，尤其是分库分表之后的逻辑表。

  • Partition

    • 主题可以被分为若⼲个分区，⼀个分区就是⼀个提交⽇志。
    • 消息以追加的⽅式写⼊分区，然后以先⼊先出的顺序读取。
    • ⽆法在整个主题范围内保证消息的顺序，但可以保证消息在单个分区内的顺序。4.Kafka通过分区来实现数据冗余和伸缩性。
    • 在需要严格保证消息的消费顺序的场景下，需要将partition数⽬设为1。

  • Replicas

    Kafka使⽤主题来组织数据，每个主题被分为若⼲个分区，每个分区有多个副本。那些副本被保存在broker上，每个broker可以保存成百上千个属于不同主题和分区的副本。

    • 首领副本

      每个分区都有⼀个⾸领副本。为了保证⼀致性，所有⽣产者请求和消费者请求都会经过这个副本

    • 跟随者副本

      ⾸领以外的副本都是跟随者副本。跟随者副本不处理来⾃客户端的请求，它们唯⼀的任务就是从⾸领那⾥复制消息，保持与⾸领⼀致的状态。如果⾸领发⽣崩溃，其中的⼀个跟随者会被提升为新⾸领。

  • Offset

    • ⽣产者Offset

      消息写⼊的时候，每⼀个分区都有⼀个offset，这个offset就是⽣产者的offset，同时也是这个分区的最新最⼤的offset。有些时候没有指定某⼀个分区的offset，这个⼯作kafka帮我们完成。

    • 消费者Offset

      这是某⼀个分区的offset情况，⽣产者写⼊的offset是最新最⼤的值是12，⽽当ConsumerA进⾏消费时，从0开始消费，⼀直消费到了9，消费者的offset就记录在9，ConsumerB就纪录在了11。等下⼀次他们再来消费时，他们可以选择接着上⼀次的位置消费，当然也可以选择从头消费，或者跳到最近的记录并从“现在”开始消费。

  • 副本

    Kafka通过副本保证⾼可⽤。副本分为⾸领副本(Leader)和跟随者副本(Follower)。

    跟随者副本包括同步副本和不同步副本，在发⽣⾸领副本切换的时候，只有同步副本可以切换为⾸领副本。

    • AR

      分区中的所有副本统称为AR(Assigned Repllicas) AR=ISR+OSR

    • ISR

      所有与leader副本保持⼀定程度同步的副本（包括Leader）组成ISR（In-SyncReplicas），ISR集合是AR集合中的⼀个⼦集。消息会先发送到leader副本，然后follower副本才能从leader副本中拉取消息进⾏同步，同步期间内follower副本相对于leader副本⽽⾔会有⼀定程度的滞后。前⾯所说的“⼀定程度”是指可以忍受的滞后范围，这个范围可以通过参数进⾏配置。

    • OSR

      与leader副本同步滞后过多的副本（不包括leader）副本，组成OSR(Out-SyncRelipcas)。在正常情况下，所有的follower副本都应该与leader副本保持⼀定程度的同步，即AR=ISR,OSR集合为空。

    • HW

      HW是High Watermak的缩写，俗称⾼⽔位，它表示了⼀个特定消息的偏移量（offset），消费之只能拉取到这个offset之前的消息。

    • LEO

      LEO是LogEndOffset的缩写，它表示了当前⽇志⽂件中下⼀条待写⼊消息的offset。

2. 安装与配置

2.1 java环境安装

安装jdk8即可, 不再做演示了, 记得配置环境变量

export JAVA_HOME=/usr/local/jdk1.8.0_231
export JRE_HOME=/usr/local/jdk1.8.0_231/jre
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib:$CLASSPATH
export PATH=$JAVA_HOME/bin:$PATH

2.2 zookeeper的安装

  • 下载稳定版zookeeper

     wget https://dlcdn.apache.org/zookeeper/zookeeper-3.7.1/apache-zookeeper-3.7.1-bin.tar.gz

  • 解压并进入conf目录,配置以下两个参数

    记得创建对应的目录

    dataDir=/tmp/zookeeper
dataLogDir=/tmp/zookeeper/logs

  • 环境变量配置

    export ZOOKEEPER_PREFIX=/home/apache-zookeeper-3.7.1-bin #根据自己的安装目录
export PATH=$PATH:$ZOOKEEPER_PREFIX/bin

  • 生效配置

    source /etc/profile

  • 验证

    zkServer.sh status

2.3 kafka的安装与配置

  • 下载

    wget https://archive.apache.org/dist/kafka/1.0.0/kafka_2.12-1.0.0.tgz

  • 解压并配置环境变量

    tar -zxf kafka_2.12-1.0.0.tgz
vim /etc/profile 

export KAFKA_HOME=/home/kafka_2.12-1.0.0 #解压目录
export PATH=$PATH:$KAFKA_HOME/bin

#使生效
source /etc/profile

  • 配置config目录下的server.properties

    #Kafka连接Zookeeper的地址，此处使⽤本地启动的Zookeeper实例，连接地址是localhost:2181，
#后⾯的myKafka 是Kafka在Zookeeper中的根节点路径
zookeeper.connect=localhost:2181/myKafka

#配置kafka连接
#内网
listeners=PLAINTEXT://192.168.0.237:9092 
#外网,此处配置外网ip才能外网连接
advertised.listeners=PLAINTEXT://114.115.171.220:9092

  • 启动zookeeper并检查状态

    zkServer.sh start
zkServer.sh status

  • 启动kafka

    # 启动命令  配置文件路径
kafka-server-start.sh kafka_2.12-1.0.0/config/server.properties
# 后台启动
kafka-server-start.sh -daemon kafka_2.12-1.0.0/config/server.properties
# 出现[KafkaServer id=0] started (kafka.server.KafkaServer)启动成功

  • 查看zookeeper的节点

2.4 生产与消费

  • kafka-topics.sh用于管理主题

    # 列出现有的主题
kafka-topics.sh --list --zookeeper localhost:2181/myKafka
# 创建主题，该主题包含⼀个分区，该分区为Leader分区，它没有Follower分区副本
kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_1 --partitions 1 --replication-factor 1
# 查看指定主题的详细信息
kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_1
# 删除指定主题
kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_1

  • kafka-console-producer.sh⽤于⽣产消息

    # 开启⽣产者
kafka-console-producer.sh --topic topic_1 --broker-list localhost:9092

  • kafka-console-consumer.sh⽤于消费消息

    # 开启消费者
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_1

# 开启消费者⽅式⼆，从头消费，不按照偏移量消费
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_1 --from-beginning

3. kafka开发实战

3.1 消息的发送与接收

生产者主要的对象有:

  • KafkaProducer

    发送消息的类

  • ProducerRecord

    封装kafka的消息

KafkaProducer创建需要指定的参数有:

参数 说明
bootstrap.servers 配置⽣产者如何与broker建⽴连接。该参数设置的是初始化参数。如果⽣产者需要连接的是Kafka集群，则这⾥配置集群中⼏个broker的地址，⽽不是全部，当⽣产者连接上此处指定的broker之后，在通过该连接发现集群中的其他节点。
key.serializer 要发送信息的key数据的序列化类。设置的时候可以写类名，也可以使⽤该类的Class对象。
value.serializer 要发送消息的alue数据的序列化类。设置的时候可以写类名，也可以使⽤该类的Class对象。
acks 默认值：all。
acks=0：
⽣产者不等待broker对消息的确认，只要将消息放到缓冲区，就认为消息已经发送完成。
该情形不能保证broker是否真的收到了消息，retries配置也不会⽣效。发送的消息的返回的消息偏移量永远是-1。
acks=1
表示消息只需要写到主分区即可，然后就响应客户端，⽽不等待副本分区的确认。
在该情形下，如果主分区收到消息确认之后就宕机了，⽽副本分区还没来得及同步该消息，则该消息丢失。
acks=all
⾸领分区会等待所有的ISR副本分区确认记录。
该处理保证了只要有⼀个ISR副本分区存活，消息就不会丢失。这是Kafka最强的可靠性保证，等效于acks=-1
retries retries重试次数
当消息发送出现错误的时候，系统会重发消息。
跟客户端收到错误时重发⼀样。
如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1 否则在重试此失败消息的时候，其他的消息可能发送成功了

其他参数可以从 org.apache.kafka.clients.producer.ProducerConfig 中找到。我们后⾯的内容会介绍到。

示例代码: kafka-demo#kafka-client-demo#demo01(https://gitee.com/ixinglan/kafka-demo.git)

3.2 Springboot Kafka

示例代码: kafka-demo#springboot-kafka-demo(https://gitee.com/ixinglan/kafka-demo.git)

4. 服务端参数配置

server.properties⽂件中的配置

4.1 zookeeper.connect

该参数⽤于配置Kafka要连接的Zookeeper/集群的地址, 例:

zookeeper.connect=node2:2181,node3:2181,node4:2181/myKafka

4.2 listeners

⽤于指定当前Broker向外发布服务的地址和端⼝

advertised.listeners 配合，⽤于做内外⽹隔离。

内外网隔离配置:

  • listener.security.protocol.map

    监听器名称和安全协议的映射配置, ⽐如，可以将内外⽹隔离，即使它们都使⽤SSL

    listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SSL

    每个监听器的名称只能在map中出现⼀次。

  • listeners

    ⽤于配置broker监听的URI以及监听器名称列表，使⽤逗号隔开多个URI及监听器名称

    如果监听器名称代表的不是安全协议，必须配置listener.security.protocol.map

    每个监听器必须使⽤不同的⽹络端⼝。

  • advertised.listeners

    需要将该地址发布到zookeeper供客户端使⽤，如果客户端使⽤的地址与listeners配置不同

    可以在zookeeper的 get /myKafka/brokers/ids/<broker.id> 中找到

    如果不设置此条⽬，就使⽤listeners的配置。跟listeners不同，该条⽬不能使⽤0.0.0.0⽹络端⼝。

    advertised.listeners的地址必须是listeners中配置的或配置的⼀部分。

  • inter.broker.listener.name

    ⽤于配置broker之间通信使⽤的监听器名称，该名称必须在advertised.listeners列表中

    如: inter.broker.listener.name=EXTERNAL

配置示例

#即名称与协议的映射关系
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
#下面INTERNAL相当于给内部管理用,EXTERNAL会通过advertised.listeners配置暴露给客户端用
listeners=INTERNAL://内网ip:9092,EXTERNAL://外网ip:9093
#暴露给zk给客户端使用
advertised.listeners=EXTERNAL://外网ip:9093
#在advertised.listeners中取
inter.broker.listener.name=EXTERNAL

4.3 broker.id

该属性⽤于唯⼀标记⼀个Kafka的Broker，它的值是⼀个任意integer值。

当Kafka以分布式集群运⾏的时候，尤为重要。

最好该值跟该Broker所在的物理主机有关的，如主机名为 host1.aaa.com ，则 broker.id=1 ，如果主机名为 192.168.100.101 ，则 broker.id=101 等等。

4.4 log.dir

通过该属性的值，指定Kafka在磁盘上保存消息的⽇志⽚段的⽬录, 它是⼀组⽤逗号分隔的本地⽂件系统路径。

如果指定了多个路径，那么broker 会根据“最少使⽤”原则，把同⼀个分区的⽇志⽚段保存到同⼀个路径下。

broker 会往拥有最少数⽬分区的路径新增分区，⽽不是往拥有最⼩磁盘空间的路径新增分区

0%