1 Kafka 简介 K afka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统(也可以当做MQ系统)，常见可以用于web/nginx日志、访问日志，消息服务等等，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

主要应用场景是：日志收集系统和消息系统。

Kafka主要设计目标如下：

1.以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。

2.高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。

3.支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。

4.同时支持离线数据处理和实时数据处理。

5.Scale out:支持在线水平扩展

2 Kafka 特性

☞ 高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。

☞ 可扩展性：kafka集群支持热扩展

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

☞ 容错性：允许集群中节点失败(若副本数量为n,则允许n-1个节点失败)

☞ 高并发：支持数千个客户端同时读写

3 Kafka 的使用场景

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

★ 消息系统：解耦和生产者和消费者、缓存消息等。

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

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

★ 流式处理：比如spark streaming和storm

4 专业术语

5 集群架构

1、架构图

Kafka集群中包含若干Producer，若干broker(Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高)，若干Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息。

2、环境准备

虚拟机准备

准备三台虚拟机用于部署zookeeper和Kafka 集群，要求硬件配置标准一样。

JDK 软件部署规划

JDK安装部署规划如下表：

Zookeeper 环境部署规划

Zookeeper集群部署前规划出消息端口、通信端口和部署目录如下：

环境部署规划

Kafka集群部署前规划出集群通讯端口、协议通信端口、控制台端口、集群名和部署目录如下：

软件部署分布

Kafka集群部署是各节点需要安装的软件分布如下：

 软件来源获取

Kafka集群部署所需要软件来源下载地址如下：

6集群实施部署

1、JDK 软件安装

分别在3台Kafka节点/opt目录下面安装JDK软件，将JDK软件包直接解压到目标安装路径下，配置好环境变量并让其变量生效成功。

到此JDK 软件安装成功！ 2、ZooKeeper 集群搭建

ZK 软件安装

分别在3台Kafka节点/opt目录下面安装zookeeper软件，将zk软件包直接解压到目标安装路径下，配置好环境变量并让其变量生效成功。

ZK 集群配置

1)配置文件修改

2)ZK 配置参数说明

3)ZK 服务创建 ServerID 标识

在三个虚拟机下的data文件夹下创建三个myid文件，并且三个文件里面分别写入1，2，3

ZK 集群启动

ZK 集群测试

模拟ZK集群切换，停掉Leader节点让ZK自动发生选举到其它节点。

上图Leader节点在192.168.58.130，其它两个节点都是follower状态。

模拟将192.168.58.130节点leader服务发生故障宕机，观察ZK集群是否会自动选举到其它两个follower节点之一作为leader节点。当leader服务主节点已经被停止后，ZK最终选举了follower节点192.168.58.131为leader，选举成功，ZK集群正常工作。后面再将宕机的那台主机恢复回来后，此节点ZK的状态就变成了被选举状态follower。至此说明ZK集群切换成功。

3、Kafka 集群搭建

Kafka集群安装采集节点克隆安装，先将kafka安装到其中一个节点，然后将配置修改后，直接将节点1上面的kafka以克隆方式COPY到其他两个节点。

kafka 软件安装

选择节点一台kafka主机服务器节点/opt目录下面安装kafka软件，安装完后并配置好环境变量并让其变量生效成功。

kafka 配置修改

修改kafka配置文件/opt/kafka-cluster/node1/config/server.properties ，注意只需要修改如下三项配置，broker.id 、listeners、zookeeper.connect 。

注意：以上kafka节点1已经配置完成。

克隆 kafka 节点

1)克隆 kafka 节点 2

将kafka节点1上面的kafka安装软件目录远程拷贝到远程节点2

2)克隆 kafka 节点 3

将kafka节点1上面的kafka安装软件目录远程拷贝到远程节点3

到此kafka 集群已经配置完成！

kafka 集群启动

别在三台kafka节点上面用命令去启停服务

● 启动 ZK 节点

/opt/zookeeper/bin/zkServer.sh start

●  停止 ZK 节点

/opt/zookeeper/bin/zkServer.sh stop

●  查看 ZK 状态

/opt/zookeeper/bin/zkServer.sh status

●  日志路径

7 Kafka   性能调优

Broker 参数配置

1、网络和io操作线程配置优化

#   broker处理消息的最大线程数(默认为3)

num.network.threads=cpu核数+1

#   broker处理磁盘IO的线程数 

num.io.threads=cpu核数*2

2、log数据文件刷盘策略   

#   每当producer写入10000条消息时，刷数据到磁盘 

log.flush.interval.messages=10000

#   每间隔1秒钟时间，刷数据到磁盘

log.flush.interval.ms=1000

3、日志保留策略配置

#   保留三天，也可以更短   (log.cleaner.delete.retention.ms)

log.retention.hours=72

#   段文件配置1GB，有利于快速回收磁盘空间，重启kafka加载也会加快(如果文件过小，则文件数量比较多，kafka启动时是单线程扫描目录(log.dir)下所有数据文件

log.segment.bytes=1073741824

4、Replica相关配置

default.replication.factor:3

#   这个参数指新创建一个topic时，默认的Replica数量,Replica过少会影响数据的可用性，太多则会白白浪费存储资源，一般建议在2~3为宜。

Java API 调优

1、zookeeper.session.timeout.ms 

解释：配置的超时时间太短，Zookeeper没有读完Consumer的数据，连接就被Consumer断开了！

参数：5000 

2、zookeeper.sync.time.ms

解释：ZooKeeper集群中leader和follower之间的同步的时间

参数：2000

3、auto.commit.enable=true 

解释：注意offset信息并不是每消费一次消息就向zk提交一次,而是现在本地保存(内存),并定期提交   

4、auto.commit.interval.ms

解释：自动提交offset到zookeeper的时间间隔

参数：1000

5、zookeeper.connection.timeout.ms 

解释：确认zookeeper连接建立操作客户端能等待的最长时间

参数：10000

6、rebalance.backoff.ms

解释：消费均衡两次重试之间的时间间隔

参数：2000

7、rebalance.max.retries

解释：消费均衡的重试次数

参数：10

8客户端配置

Producer 客户端配置

 Consumer 客户端配置

本期作者|晏杰宏  主要负责数据库、中间件、大数据等基础软件建设、优化和业务保障工作。具有10年的电信与银行企业一线/二线运维服务管理经验。目前专注于云计算、中间件、开源技术和数据库等领域技术研究。持有Oracle OCP、weblogic OCP、Docker容器、Postgresql PGCE和阿里云ACP等认证。