Windows环境下Kafka与Zookeeper安装配置与启动过程

【Kafka源码】Kafka启动过程⼀般来说，我们是通过命令来启动kafka，但是命令的本质还是调⽤代码中的main⽅法，所以，我们重点看下启动类Kafka。

源码下下来之后，我们也可以通过直接运⾏Kafka.scala中的main⽅法（需要指定启动参数，也就是server.properties的位置）来启动Kafka。

因为kafka依赖zookeeper，所以我们需要提前启动zookeeper，然后在server.properties中指定zk地址后，启动。

下⾯我们⾸先看⼀下main()⽅法：def main(args: Array[String]): Unit = {try {val serverProps = getPropsFromArgs(args)val kafkaServerStartable = KafkaServerStartable.fromProps(serverProps)// attach shutdown handler to catch control-cRuntime.getRuntime().addShutdownHook(new Thread() {override def run() = {kafkaServerStartable.shutdown}})kafkaServerStartable.startupkafkaServerStartable.awaitShutdown}catch {case e: Throwable =>fatal(e)System.exit(1)}System.exit(0)}我们慢慢来分析下，⾸先是getPropsFromArgs(args)，这⼀⾏很明确，就是从配置⽂件中读取我们配置的内容，然后赋值给serverProps。

第⼆步，KafkaServerStartable.fromProps(serverProps)，object KafkaServerStartable {def fromProps(serverProps: Properties) = {KafkaMetricsReporter.startReporters(new VerifiableProperties(serverProps))new KafkaServerStartable(KafkaConfig.fromProps(serverProps))}}这块主要是启动了⼀个内部的监控服务（内部状态监控）。

Kafka安装及配置过程⼀.安装2.Kafka与scala版本也有关系。

3.kafka安装之前需要提前配置好hadoop集群与zookeeper集群。

kafka是依赖于zookeeper集群的。

hadoop集群：master，slave1，slave24.解压：tar -zxvf kafka_2.12-2.5.1.tgz -C /usr/hadoop⼆.修改配置⽂件进⼊解压⽂件 /usr/hadoop/kafka_2.12-2.5.1/config 这个⽬录下1.修改 server.properties#broker的全局唯⼀编号，不能重复broker.id=0#⽤来监听链接的端⼝，producer或consumer将在此端⼝建⽴连接port=9092#处理⽹络请求的线程数量work.threads=3#⽤来处理磁盘IO的线程数量num.io.threads=8#发送套接字的缓冲区⼤⼩socket.send.buffer.bytes=102400#接受套接字的缓冲区⼤⼩socket.receive.buffer.bytes=102400#请求套接字的缓冲区⼤⼩socket.request.max.bytes=104857600#kafka消息存放的路径log.dirs=/usr/hadoop/kafka_2.12-2.5.1/data#topic在当前broker上的分⽚个数num.partitions=2#⽤来恢复和清理data下数据的线程数量num.recovery.threads.per.data.dir=1#segment⽂件保留的最长时间，超时将被删除log.retention.hours=168#滚动⽣成新的segment⽂件的最⼤时间log.roll.hours=168#⽇志⽂件中每个segment的⼤⼩，默认为1Glog.segment.bytes=1073741824#周期性检查⽂件⼤⼩的时间log.retention.check.interval.ms=300000#⽇志清理是否打开log.cleaner.enable=true#broker需要使⽤zookeeper保存meta数据zookeeper.connect=master:2181,slave1:2181,slave2:2181#zookeeper链接超时时间zookeeper.connection.timeout.ms=6000#partion buffer中，消息的条数达到阈值，将触发flush到磁盘log.flush.interval.messages=10000#消息buffer的时间，达到阈值，将触发flush到磁盘log.flush.interval.ms=3000#删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除delete.topic.enable=true#此处的为本机IP(重要),如果不改,则客户端会抛出:Producerconnection to localhost:9092 unsuccessful 错误!=master注：broker.id=0 这个配置在每个配置节点上要修改不同的值。

Window下zookeeper安装和启动请参考和1.下载稳定的最新版本的zookeeper，这⾥下载了zookeeper-3.5.7版本注意 apache-zookeeper-3.5.7-bin.tar.gz 才是启动zookeepr需要的，千万别下载错。

2.解压缩 apache-zookeeper-3.5.7-bin.tar.gz 。

在该⽂件夹下打开cmd命令，使⽤如下命令解压缩⽂件tar -zxvf apache-zookeeper-3.7.7-bin.tar.gz3.复制/conf ⽂件夹下的zoo_sample.cfg 并重命名为zoo.cfg。

修改zoo.cfg⽂件如下(配置log路径)添加如下两⾏三⾏设置log⽂件路径并修改zookeeper的启动端⼝为8082，默认为8080dataDir=C:\\zookeeper\\apache-zookeeper-3.5.7-bin\\datadataLogDir=C:\\zookeeper\\apache-zookeeper-3.5.7-bin\\logadmin.serverPort=8082完整⽂件如下# The number of milliseconds of each ticktickTime=2000# The number of ticks that the initial# synchronization phase can takeinitLimit=10# The number of ticks that can pass between# sending a request and getting an acknowledgementsyncLimit=5# the directory where the snapshot is stored.# do not use /tmp for storage, /tmp here is just# example sakes.dataDir=C:\\zookeeper\\apache-zookeeper-3.5.7-bin\\datadataLogDir=C:\\zookeeper\\apache-zookeeper-3.5.7-bin\\log# the port at which the clients will connectclientPort=2181# the maximum number of client connections.# increase this if you need to handle more clients#maxClientCnxns=60## Be sure to read the maintenance section of the# administrator guide before turning on autopurge.## /doc/current/zookeeperAdmin.html#sc_maintenance## The number of snapshots to retain in dataDir#autopurge.snapRetainCount=3# Purge task interval in hours# Set to "0" to disable auto purge feature#autopurge.purgeInterval=1#change server port, 8080 by defaultadmin.serverPort=80824.修改 bin\zkEnv.cmd⽂件，配置JAVA_HOME，否则启动zookeepr会报错参考添加set JAVA_HOME="C:\java\jdk1.8.0_212-X64"修改后的部分zkEnv.cmd⽂件如下@REM setup java environment variablesset JAVA_HOME="C:\java\jdk1.8.0_212-X64"if not defined JAVA_HOME (echo Error: JAVA_HOME is not set.goto :eof)set JAVA_HOME=%JAVA_HOME:"=%if not exist "%JAVA_HOME%"\bin\java.exe (echo Error: JAVA_HOME is incorrectly set: %JAVA_HOME%echo Expected to find java.exe here: %JAVA_HOME%\bin\java.exegoto :eof)5.启动zookeeper，在/bin⽬录下打开cmd窗⼝，并执⾏命令 zkServer.cmd。

kafka入门教程
Kafka是一个分布式流处理平台，被广泛应用于大数据处理和
实时数据流处理的场景。

它具有高吞吐量、可扩展性强、持久性和容错能力、以及多种数据处理模式等特点。

本教程将引导你进入Kafka的世界，教你如何安装、配置和使
用Kafka。

下面是教程内容概述：
1. 下载和安装Kafka
首先，你需要从Kafka的官方网站上下载最新版本的Kafka。

接下来，我们将介绍如何在不同操作系统上进行安装和配置。

2. 配置Kafka
在安装完成后，你需要对Kafka进行一些必要的配置。

这包
括设置Zookeeper连接、主题（Topic）的配置、分区（Partition）的配置等。

3. 创建生产者和消费者
一旦配置完成，你就可以创建Kafka的生产者和消费者实例了。

生产者可将消息发送到指定的主题，消费者则可以订阅特定的主题并消费其中的消息。

4. 发送和接收消息
通过创建生产者并将消息发送至主题，你可以实现向Kafka
集群发送消息的功能。

消费者可以通过订阅主题并接收消息来完成消息消费的功能。

5. Kafka的高级特性
在熟悉基本使用后，你还可以学习一些更高级的Kafka特性，如消息分区、消息持久化、消息的顺序性等。

6. 故障处理和调优
正确处理故障和进行性能优化是使用Kafka的关键。

我们将
介绍一些常见的故障处理和性能优化技巧。

通过本教程，你将能够迅速上手使用Kafka，并了解它的基本
概念和常用功能。

希望这对你入门Kafka有所帮助！。

Zookeeper配置开机启动ZooKeeper是Hadoop的正式子项目；Hadoop是一个分布式系统基础架构，由Apache基金会所开发；Zookeeper能够用来leader选举；也就是你有N+1台同样的服务器的时候又zookeeper来决定谁是主服务器；当我们配置好zookeeper之后都希望当系统重启之后能够自动启动zookeeper程序；今天小编就为大家介绍如何得配置zookeeper的开机启动本次演示环境是在Centos7下完成的；其他版本的centos都是一样的操作步骤如下：1.请自行下载安装配置zookeeper的服务器环境本经验仅仅介绍如何配置zookeeper的开机自动启动2.首先请登陆你的linux服务器，用cd 命令切换到/etc/rc.d/init.d/目录下3.接着用touch zookeeper创建一个文件4.然后为这个文件添加可执行权限chmod +x zookeeper5.接着用vi zookeeper来编辑这个文件6.接着在zookeeper里面输入如下内容（路径根据实际情况编写）#!/bin/bash#chkconfig:2345 20 90#description:zookeeper#processname:zookeepercase $1 instart) su root /usr/local/sw/zookeeper/bin/zkServer.sh start;;stop) su root /usr/local/sw/zookeeper/bin/zkServer.sh stop;;status) su root /usr/local/sw/zookeeper/bin/zkServer.sh status;;restart) su root /usr/local/sw/zookeeper/bin/zkServer.sh restart;;*) echo "require start|stop|status|restart" ;;esac输入:wq 保存退出7.这个时候我们就可以用service zookeeper start/stop来启动停止zookeeper服务了8.最后一点我们需要开机自动启动，所以需要添加到启动里面。

kafka zookeeper篇(组件、原理、使用场景、面试)Kafka与Zookeeper的结合：1. Kafka是一个开源流处理平台，提供流数据生产和消费的服务。

它主要用于构建实时数据流管道和应用，如日志收集、消息分发、流数据处理等。

2. Zookeeper是一个分布式协调服务，提供分布式应用程序的数据同步、配置管理和服务发现等功能。

Kafka依赖于Zookeeper来管理集群的元数据信息和协调集群中的broker节点。

Kafka与Zookeeper的结合原理：1. Kafka集群中的每个broker节点都会将自己注册到Zookeeper中，并保存自身的元数据信息，如配置信息和broker状态等。

2. Zookeeper保存了整个Kafka集群的元数据信息，包括Topic信息、Partition信息和Broker信息等。

这些信息对Kafka的分布式协调非常重要，可以确保Kafka 集群的正常运行。

3. 当Kafka客户端进行生产或消费操作时，它会首先与Zookeeper进行交互，获取集群的状态信息和元数据信息，然后根据这些信息进行相应的操作。

Kafka与Zookeeper的使用场景：1. Kafka在实时数据处理领域广泛应用，如日志收集、消息分发和流数据处理等。

它可以处理大量的数据流，并提供高吞吐量和低延迟的服务。

2. Zookeeper常用于分布式系统中的协调和服务发现。

它可以管理分布式节点的状态信息和元数据信息，提供一致性和可靠性保证。

Kafka与Zookeeper的面试问题：1. Kafka是什么？它有哪些特点？2. Zookeeper是什么？它有哪些功能？3. Kafka与Zookeeper的关系是什么？它们是如何结合的？4. Kafka如何保证数据的安全性和可靠性？5. Zookeeper在Kafka中的作用是什么？它如何管理Kafka集群的元数据信息？。

kafka在windows下的安装和配置博主最近在学习有关kafka的配置安装以及在spring的集成使⽤。

但⽹上关于kafka的配置参考资料基本都是于linux下的配置，于是博主在整理了相关windows下kafka的配置记录在博客⾥。

由于是简单配置所以在这⾥只建了⼀个topic以及⼀个producer和两个consumer。

在官⽹上下载 zookeeper和kafka(我下的版本kafka_2.11-0.11.0.0，这个版本中bin⽬录下有windows⽬录)，注意不要下载源码包（名字中带有src），否则启动的时候会报错。

1、配置好jdk环境2、解压zookeeper到指定⽬录，找到解压后⽬录中conf⽂件夹中zoo_sample - 副本.cfg⽂件，复制在conf中改名为zoo.cfg。

在bin⽂件夹中打开zkServer.bat启动zookeeper。

⾄此，zookeeper启动完成。

3、解压kafka到指定⽬录。

查看kafka根⽬录中config⽂件夹下server.properties，确认其中关于zookeeper的连接端⼝和zookeep中zoo.cfg的端⼝⼀致。

3.1、启动kafka 在cmd中进⼊kafka根⽬录。

输⼊以下命令： .\bin\windows\kafka-server-start.bat .\config\server.properties kafka启动成功 3.2、创建topic 在cmd中进⼊kafka\bin\windows⽬录，输⼊以下命令： kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test　 创建成功 3.3、创建kafka producer 在cmd中进⼊kafka\bin\windows⽬录，输⼊以下命令： kafka-console-producer.bat --broker-list localhost:9092 --topic test 注意这⾥的端⼝和server.properties的端⼝号⼀致。

kafka安装及配置过程⼀、安装kafka根据Scala版本不同，⼜分为多个版本，我不需要使⽤Scala，所以就下载官⽅推荐版本kafka_2.12-2.4.0.tgz。

使⽤tar -xzvf kafka_2.12-2.4.0.tgz 解压为了使⽤⽅便，可以创建软链接kafka0⼆、Zookeeper配置当前下载的kafka程序⾥⾃带Zookeeper，可以直接使⽤其⾃带的Zookeeper建⽴集群，也可以单独使⽤Zookeeper安装⽂件建⽴集群。

1. 单独使⽤Zookeeper安装⽂件建⽴集群Zookeeper的安装及配置可以参考另⼀篇博客，⾥⾯有详细介绍2. 直接使⽤其⾃带的Zookeeper建⽴集群kafka⾃带的Zookeeper程序脚本与配置⽂件名与原⽣Zookeeper稍有不同。

kafka⾃带的Zookeeper程序使⽤bin/zookeeper-server-start.sh，以及bin/zookeeper-server-stop.sh来启动和停⽌Zookeeper。

⽽Zookeeper的配制⽂件是config/zookeeper.properties，可以修改其中的参数（1）启动Zookeeperbin/zookeeper-server-start.sh -daemon config/zookeeper.properties加-daemon参数，可以在后台启动Zookeeper，输出的信息在保存在执⾏⽬录的logs/zookeeper.out⽂件中。

对于⼩内存的服务器，启动时有可能会出现如下错误os::commit_memory(0x00000000e0000000, 536870912, 0) failed; error='Not enough space' (errno=12)可以通过修改bin/zookeeper-server-start.sh中的参数，来减少内存的使⽤，将下图中的-Xmx512M -Xms512M改⼩。

Ubuntu18.04设置zookeeper和kafka开机自启动1 进入/etc/init.d/目录cd /etc/init.d/2 创建zookeeper文件touch zookeeper3 赋予文件权限sudo chmod +x zookeeper4命令写入执行脚本vim zookeeper#! /bin/sh### BEGIN INIT INFO# Short-Description: Zookeeper# Description: Zookeeper# Provides: flex# Required-Start: $local_fs $network# Required-Stop: $local_fs# Default-Start: 2 3 4 5# Default-Stop: 0 1 6### END INIT INFOcase "$1" instart)su -l root -c ' /usr/local/hcj/zookeeper-3.3.4/bin/zkServer.sh start'exit 0;;stop)su -l root -c ' /usr/local/hcj/zookeeper-3.3.4/bin/zkServer.sh stop' exit 0;;*) echo 'require start|stop'exit 1;;esac5重新载入配置文件systemctl reload zookeeper6设置启动开机自启动systemctl enable zookeeper7重启reboot8查看是否启动成功sudo jps9 kafka自启动vim /etc/rc-local#添加以下启动脚本cd /home/hcj/kafka/kafka_2.11-1.0.2bash bin/kafka-server-start.sh -daemon config/server.properties bash bin/kafka-server-start.sh -daemon config/server1.properties bash bin/kafka-server-start.sh -daemon config/server2.properties如果启动不成功修改启动文件配置vim kafka-run-class.sh将JAVA_MAJOR_VERSION=$($JAVA -version 2>&1 | sed -E -n 's/.* version "([^.-]*).*"/\1/p')修改为JAVA_MAJOR_VERSION=$($JAVA -version 2>&1 | sed -E -n 's/.* version "([^.-]*).*/\1/p')[Unit]# Zookeeper服务的描述Description=Zookeeper Service# 服务依赖—在什么服务之后启动，一般为在网络服务启动后启动After=network.target[Service]# 服务类型—如果是shell脚本的方式,则Type=forking,否则不指定作何值(也就是去掉该配置项)Type=forking# 启动环境参数# 此脚本指定了Zookeeper日志和Java的目录Environment=ZOO_LOG_DIR=/usr/local/hcj/zookeeper-3.3.4/log/Environment=JAVA_HOME=/usr/local/hcj/jdk1.8.0_231# 启动命令ExecStart=/usr/local/hcj/zookeeper-3.3.4/bin/zkServer.sh start# 停止命令ExecStop=/usr/local/hcj/zookeeper-3.3.4/bin/zkServer.sh stop# 重启命令ExecReload=/usr/local/hcj/zookeeper-3.3.4/bin/zkServer.sh restart[Install]WantedBy=multi-user.target。