Kafka的基本介绍和在linux的安装配置

1.介绍

Kafka最初是由LinkedIn公司采用Scala语言开发的一个多分区、多副本并且基于ZooKeeper协调的分布式消息系统，现在已经捐献给了Apache基金会。目前Kafka已经定位为一个分布式流式处理平台，它以高吞吐、可持久化、可水平扩展、支持流处理等多种特性而被广泛应用。

Apache Kafka是一个分布式的发布-订阅消息系统，能够支撑海量数据的数据传递。在离线和实时的消息处理业务系统中，Kafka都有广泛的应用。Kafka将消息持久化到磁盘中，并对消息创建了备份保证了数据的安全。Kafka在保证了较高的处理速度的同时，又能保证数据处理的低延迟和数据的零丢失。

2.特性

（1）高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个主题可以分多个分区, 消费组对分区进行消费操作；

（2）可扩展性：kafka集群支持热扩展；

（3）持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失；

（4）容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）；

（5）高并发：支持数千个客户端同时读写；

（6）日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如Hadoop、Hbase、Solr等；

（7）消息系统：解耦和生产者和消费者、缓存消息等；

（8）用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到Hadoop、数据仓库中做离线分析和挖掘；

（9）运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告；

（10）流式处理：比如spark streaming和storm；

3.技术优势

可伸缩性：Kafka 的两个重要特性造就了它的可伸缩性。
- Kafka 集群在运行期间可以轻松地扩展或收缩（可以添加或删除代理），而不会宕机
- 可以扩展一个 Kafka 主题来包含更多的分区。由于一个分区无法扩展到多个代理，所以它的容量受到代理磁盘空间的限制。能够增加分区和代理的数量意味着单个主题可以存储的数据量是没有限制的
容错性和可靠性：Kafka 的设计方式使某个代理的故障能够被集群中的其他代理检测到。由于每个主题都可以在多个代理上复制，所以集群可以在不中断服务的情况下从此类故障中恢复并继续运行。
吞吐量：代理能够以超快的速度有效地存储和检索数据。

4.概念详解

1）Producer

生产者即数据的发布者，该角色将消息发布到Kafka的topic中。broker接收到生产者发送的消息后，broker将该消息追加到当前用于追加数据的segment文件中。生产者发送的消息，存储到一个partition中，生产者也可以指定数据存储的partition。

2）Consumer

消费者可以从broker中读取数据。消费者可以消费多个topic中的数据。

3）Topic

在Kafka中，使用一个类别属性来划分数据的所属类，划分数据的这个类称为topic。如果把Kafka看做一个数据库，topic可以理解为数据库中的一张表，topic的名字即为表名。

4）Partition

topic中的数据分割为一个或多个partition。每个topic至少有一个partition。每个partition中的数据使用多个segment文件存储。partition中的数据是有序的，partition间的数据丢失了数据的顺序。如果topic有多个partition，消费数据时就不能保证数据的顺序。在需要严格保证消息的消费顺序的场景下，需要将partition数目设为1。

5）Partition offset

每条消息都有一个当前Partition下唯一的64字节的offset，它指明了这条消息的起始位置。

6）Replicas of partition

副本是一个分区的备份。副本不会被消费者消费，副本只用于防止数据丢失，即消费者不从为follower的partition中消费数据，而是从为leader的partition中读取数据。副本之间是一主多从的关系。

7）Broker

Kafka 集群包含一个或多个服务器，服务器节点称为broker。broker存储topic的数据。如果某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集群数据不均衡。

8）Leader

9）Follower

跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower，Follower与Leader保持数据同步。如果Leader失效，则从Follower中选举出一个新的Leader。当Follower与Leader挂掉、卡住或者同步太慢，leader会把这个follower从“in sync replicas”（ISR）列表中删除，重新创建一个Follower。

10）Zookeeper

Zookeeper负责维护和协调broker。当Kafka系统中新增了broker或者某个broker发生故障失效时，由ZooKeeper通知生产者和消费者。生产者和消费者依据Zookeeper的broker状态信息与broker协调数据的发布和订阅任务。

11）AR(Assigned Replicas)

分区中所有的副本统称为AR。

12）ISR(In-Sync Replicas)

所有与Leader部分保持一定程度的副本（包括Leader副本在内）组成ISR。

13）OSR(Out-of-Sync-Replicas)

与Leader副本同步滞后过多的副本。

14）HW(High Watermark)

高水位，标识了一个特定的offset，消费者只能拉取到这个offset之前的消息。

15）LEO(Log End Offset)

即日志末端位移(log end offset)，记录了该副本底层日志(log)中下一条消息的位移值。注意是下一条消息！也就是说，如果LEO=10，那么表示该副本保存了10条消息，位移值范围是[0, 9]。

5.安装与配置

1）首先需要安装Java环境

yum install java-1.8.0-openjdk* -y

2）安装zookeeper

Zookeeper是安装Kafka集群的必要组件，Kafka通过Zookeeper来实施对元数据信息的管理，包括集群、主题、分区等内容。同样在官网下载安装包到指定目录解压缩，步骤如下： ZooKeeper官网：http://zookeeper.apache.org/

tar -zxvf zookeeper-3.4.10.tar.gz
mv zookeeper-3.4.10 zookeeper

修改Zookeeper的配置文件，首先进入安装路径conf目录，并将zoo_sample.cfg文件修改为 zoo.cfg，并对核心参数进行配置。

dataDir=/tmp/zookeeper/data
dataLogDir=/tmp/zookeeper/log

启动Zookeeper命令：

bin/zkServer.sh start

ZooKeeper JMX enabled by default
Using config: /data/soft/zookeeper/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

3）安装Kafka

官网下载安装解压缩：http://kafka.apache.org/downloads

tar -zxvf kafka_2.12-2.2.1
mv kafka_2.12-2.2.1 kafka

下载解压启动

启动命令

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

server.properties配置中需要关注以下几个参数：

broker.id=0 表示broker的编号，如果集群中有多个broker，则每个broker的编号需要设置的不同

listeners=PLAINTEXT://:9092 brokder对外提供的服务入口地址

log.dirs=/tmp/kafka/log 设置存放消息日志文件的地址

zookeeper.connect=localhost:2181 Kafka所需Zookeeper集群地址

启动成功如下显示：

启动成功之后重新打开一个终端，验证启动进程

4）Kafka测试消息生产和消费

首先创建一个主题

bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic test --partitions 2 --replication-factor 1


--zookeeper：指定了Kafka所连接的Zookeeper服务地址
--topic：指定了所要创建主题的名称
--partitions：指定了分区个数
--replication-factor：指定了副本因子
--create：创建主题的动作指令

展示所有主题

bin/kafka-topics.sh --zookeeper localhost:2181 --list

查看主题详情

bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic test


--describe 查看详情动作指令
--topic 指定了消费端订阅的主题

生产端发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

--broker-list 指定了连接的Kafka集群的地址
--topic 指定了发送消息时的主题

消费端消费消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test

--bootstrap-server 指定了连接Kafka集群的地址
--topic 指定了消费端订阅的主题

5）常用参数配置

zookeeper.connect

指明Zookeeper主机地址，如果zookeeper是集群则以逗号隔开，

如：172.6.14.61:2181,172.6.14.62:2181,172.6.14.63:2181

listeners

监听列表，broker对外提供服务时绑定的IP和端口。多个以逗号隔开，如果监听器名称不是一个安全的协议， listener.security.protocol.map也必须设置。主机名称设置0.0.0.0绑定所有的接口，主机名称为空则绑定默认的接口。

如：PLAINTEXT://myhost:9092,SSL://:9091 CLIENT://0.0.0.0:9092,REPLICATION://localhost:9093

broker.id

broker的唯一标识符，如果不配置则自动生成，建议配置且一定要保证集群中必须唯一，默认-1

log.dirs

日志数据存放的目录，如果没有配置则使用log.dir

message.max.bytes

服务器接受单个消息的最大大小，默认1000012 约等于976.6KB。

log.retention.{hours|minutes|ms}

Kafka 通常根据时间来决定数据可以被保留多久。默认使用 log.retention.hour 参数来配置时间，默认值为 168 小时，也就是一周。除此以外，还有其他两个参数 minutes|ms。

虽然 ms 设置有最高的优先级，但是通常情况下我们还是设置 hours 级别的多一些，比如log.retention.hours=168表示默认保存 7 天的数据，自动删除 7 天前的数据。很多公司把 Kafka 当做存储来使用，那么这个值就要相应地调大。

log.retention.bytes

这是指定 Broker 为消息保存的总磁盘容量大小。这个值默认是 -1，表明你想在这台 Broker 上保存多少数据都可以，至少在容量方面 Broker 绝对为你开绿灯，不会做任何阻拦。这个参数真正发挥作用的场景其实是在云上构建多租户的 Kafka 集群：设想你要做一个云上的 Kafka 服务，每个租户只能使用 100GB 的磁盘空间，为了避免有个“恶意”租户使用过多的磁盘空间，设置这个参数就显得至关重要了。

如果同时指定了log.retention.xx和log.retention.bytes，只要任意一个条件得到满足，消息就会被删除。例如log.retention.hour=24(也就是1天)，log.retention.bytes=1 000 000 000(也就是1G)，那么消息总数在1天内超过1G的部分将会被删除。相反，如果消息总数小于1G，1天之后也会被删除，尽管分区的数据总量小于1GB。

log.segment.bytes

以上的设置都作用在日志片段上，而不是作用在单个消息上。当消息到达 broker 时，它们被迫加到分区的当前日志片段上。当日志片段大小达到 log.segment bytes 定的上限（默认是 lGB ）时，当前日志片段就会被关闭，一个新的日志片段被打开。如果一个日志片段被关闭，就开始等待过期。这个参数的值越小，就会越频繁地关闭和分配新文件，从而降低磁盘入的整体效率。

log.segment.ms

它指定了多长时间之后日志片段会被关闭。就像 log.retention.ms和log.retention.bytes这两个参数样， log.segment.ms和log.segment.bytes 这两个参数之间也不存在互斥问题。日志片段会在大小或时间达到上限时被关闭，就看哪个条件先得到满足。

message.max.bytes

控制 Broker 能够接收的单个消息的最大大小。这个参数也是一样，默认的 1000 000 太少了，还不到 1MB。实际场景中突破 1MB 的消息都是屡见不鲜的，因此在线上环境中设置一个比较大的值还是比较保险的做法。毕竟它只是一个标尺而已，仅仅衡量 Broker 能够处理的最大消息大小，即使设置大一点也不会耗费什么磁盘空间的。