Windows安装kafka

压缩包下载地址：https://www.apache.org/dyn/closer.cgi?path=/kafka/3.6.1/kafka_2.13-3.6.1.tgz

启动kafka步骤

zookeeper-server-start.bat

rem 关闭命令提示符窗口的命令回显，这样在运行脚本时不会显示脚本的具体命令内容
@echo off
rem 命令行启动未提供参数则输出: USAGE:命令 zookeeper.properties 中止
IF [%1] EQU [] (
	echo USAGE: %0 zookeeper.properties
	EXIT /B 1
)
rem 开始了一个局部环境，只在当前脚本执行过程中有效
SetLocal
rem 未设置值，则设置
IF ["%KAFKA_LOG4J_OPTS%"] EQU [""] (
    set KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:%~dp0../../config/log4j.properties
)
IF ["%KAFKA_HEAP_OPTS%"] EQU [""] (
    set KAFKA_HEAP_OPTS=-Xmx512M -Xms512M
)
rem %~dp0获取命令的文件路径，%* 读取执行这个命令携带的锁有参数
"%~dp0kafka-run-class.bat" org.apache.zookeeper.server.quorum.QuorumPeerMain %*
rem 在脚本中限定变量的作用范围，以防止对环境变量的更改影响到脚本之外的其他部分
EndLocal

kafka-run-class.bat

@echo off
setlocal enabledelayedexpansion
IF [%1] EQU [] (
	echo USAGE: %0 classname [opts]
	EXIT /B 1
)
rem Using pushd popd to set BASE_DIR to the absolute path
rem pushd表示切换目录,这里作用是切换到%~dp0对应的上2级目录
pushd %~dp0..\..
rem CD是系统变量,%CD%表示当前工作目录的路径
set BASE_DIR=%CD%
rem popd 切回原来的目录
popd
IF ["%SCALA_VERSION%"] EQU [""] (
  set SCALA_VERSION=2.13.11
)
rem 延迟环境变量扩展后的一种变量访问方式!var!
IF ["%SCALA_BINARY_VERSION%"] EQU [""] (
  for /f "tokens=1,2 delims=." %%a in ("%SCALA_VERSION%") do (
    set FIRST=%%a
    set SECOND=%%b
    if ["!SECOND!"] EQU [""] (
      set SCALA_BINARY_VERSION=!FIRST!
    ) else (
      set SCALA_BINARY_VERSION=!FIRST!.!SECOND!
    )
  )
)
rem Classpath addition for kafka-core dependencies 为kafka核心依赖添加类路径
rem for %%i in (...) do (...)这是一个 for 循环，用于遍历指定路径下满足特定条件的文件
for %%i in ("%BASE_DIR%\core\build\dependant-libs-%SCALA_VERSION%\*.jar") do (
	rem 
	call :concat "%%i"
)
rem Classpath addition for kafka-examples
for %%i in ("%BASE_DIR%\examples\build\libs\kafka-examples*.jar") do (
	call :concat "%%i"
)
rem Classpath addition for kafka-clients
for %%i in ("%BASE_DIR%\clients\build\libs\kafka-clients*.jar") do (
	call :concat "%%i"
)
rem Classpath addition for kafka-streams
for %%i in ("%BASE_DIR%\streams\build\libs\kafka-streams*.jar") do (
	call :concat "%%i"
)
rem Classpath addition for kafka-streams-examples
for %%i in ("%BASE_DIR%\streams\examples\build\libs\kafka-streams-examples*.jar") do (
	call :concat "%%i"
)
for %%i in ("%BASE_DIR%\streams\build\dependant-libs-%SCALA_VERSION%\rocksdb*.jar") do (
	call :concat "%%i"
)
rem Classpath addition for kafka tools 为kafka工具添加类路径
for %%i in ("%BASE_DIR%\tools\build\libs\kafka-tools*.jar") do (
	call :concat "%%i"
)
for %%i in ("%BASE_DIR%\tools\build\dependant-libs-%SCALA_VERSION%\*.jar") do (
	call :concat "%%i"
)
rem 两层for循环
for %%p in (api runtime file json tools) do (
	for %%i in ("%BASE_DIR%\connect\%%p\build\libs\connect-%%p*.jar") do (
		call :concat "%%i"
	)
	if exist "%BASE_DIR%\connect\%%p\build\dependant-libs\*" (
		call :concat "%BASE_DIR%\connect\%%p\build\dependant-libs\*"
	)
)
rem Classpath addition for release
for %%i in ("%BASE_DIR%\libs\*") do (
	call :concat "%%i"
)
rem Classpath addition for core
for %%i in ("%BASE_DIR%\core\build\libs\kafka_%SCALA_BINARY_VERSION%*.jar") do (
	call :concat "%%i"
)
rem JMX settings JMX的设置
IF ["%KAFKA_JMX_OPTS%"] EQU [""] (
	set KAFKA_JMX_OPTS=-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false
)
rem JMX port to use 使用的JMX端口
IF ["%JMX_PORT%"] NEQ [""] (
	set KAFKA_JMX_OPTS=%KAFKA_JMX_OPTS% -Dcom.sun.management.jmxremote.port=%JMX_PORT%
)
rem Log directory to use 要使用的日志目录
IF ["%LOG_DIR%"] EQU [""] (
    set LOG_DIR=%BASE_DIR%/logs
)
rem Log4j settings
IF ["%KAFKA_LOG4J_OPTS%"] EQU [""] (
	set KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:%BASE_DIR%/config/tools-log4j.properties
) ELSE (
  rem create logs directory 创建一个目录文件夹
  IF not exist "%LOG_DIR%" (
      mkdir "%LOG_DIR%"
  )
)
set KAFKA_LOG4J_OPTS=-Dkafka.logs.dir="%LOG_DIR%" "%KAFKA_LOG4J_OPTS%"
rem Generic jvm settings you want to add 要添加的通用jvm设置
IF ["%KAFKA_OPTS%"] EQU [""] (
	set KAFKA_OPTS=
)
set DEFAULT_JAVA_DEBUG_PORT=5005
set DEFAULT_DEBUG_SUSPEND_FLAG=n
rem Set Debug options if enabled
IF ["%KAFKA_DEBUG%"] NEQ [""] (
	IF ["%JAVA_DEBUG_PORT%"] EQU [""] (
		set JAVA_DEBUG_PORT=%DEFAULT_JAVA_DEBUG_PORT%
	)
	IF ["%DEBUG_SUSPEND_FLAG%"] EQU [""] (
		set DEBUG_SUSPEND_FLAG=%DEFAULT_DEBUG_SUSPEND_FLAG%
	)
	set DEFAULT_JAVA_DEBUG_OPTS=-agentlib:jdwp=transport=dt_socket,server=y,suspend=!DEBUG_SUSPEND_FLAG!,address=!JAVA_DEBUG_PORT!
	IF ["%JAVA_DEBUG_OPTS%"] EQU [""] (
		set JAVA_DEBUG_OPTS=!DEFAULT_JAVA_DEBUG_OPTS!
	)
	echo Enabling Java debug options: !JAVA_DEBUG_OPTS!
	set KAFKA_OPTS=!JAVA_DEBUG_OPTS! !KAFKA_OPTS!
)
rem Which java to use
IF ["%JAVA_HOME%"] EQU [""] (
	set JAVA=java
) ELSE (
	set JAVA="%JAVA_HOME%/bin/java"
)
rem Memory options
IF ["%KAFKA_HEAP_OPTS%"] EQU [""] (
	set KAFKA_HEAP_OPTS=-Xmx256M
)
rem JVM performance options
IF ["%KAFKA_JVM_PERFORMANCE_OPTS%"] EQU [""] (
	set KAFKA_JVM_PERFORMANCE_OPTS=-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true
)
IF not defined CLASSPATH (
	echo Classpath is empty. Please build the project first e.g. by running 'gradlew jarAll'
	EXIT /B 2
)
rem %?% 读取环境变量 
set COMMAND=%JAVA% %KAFKA_HEAP_OPTS% %KAFKA_JVM_PERFORMANCE_OPTS% %KAFKA_JMX_OPTS% %KAFKA_LOG4J_OPTS% -cp "%CLASSPATH%" %KAFKA_OPTS% %*
rem echo.
rem echo %COMMAND%
rem echo.
%COMMAND%
rem 将控制权转移到脚本的文件尾
goto :eof
:concat
rem 这是一个条件语句，用于检查CLASSPATH环境变量是否已定义
IF not defined CLASSPATH (
  set CLASSPATH="%~1"
) ELSE (
  set CLASSPATH=%CLASSPATH%;"%~1"
)

启动失败日志

缺少启动参数

输入行太长

组合起来的执行命令太长，Windows系统不支持。将项目放到盘符目录下。

找不到类

为了解决路径太长，去掉了一级目录。因为启动命令中有通过当前命令的位置去加载jar包。

绝对路径配置文件加载报错

# 文件地址错误
/kafka_2.13-3.6.1/bin/windows/zookeeper-server-start.bat /kafka_2.13-3.6.1/config/zookeeper.properties

先对路径无法启动

./bin/windows/zookeeper-server-start.bat ./bin/windows/zookeeper.properties
/bin/windows/zookeeper-server-start.bat /bin/windows/zookeeper.properties
bin/windows/zookeeper-server-start.bat bin/windows/zookeeper.properties

启动成功

1. 启动zookeeper

# 当前路径下启动
zookeeper-server-start.bat ../../config/zookeeper.properties
# 将文件复制一份到启动命令处
zookeeper-server-start.bat zookeeper.properties
# 绝对路径启动
/kafka_2.13-3.6.1/bin/windows/zookeeper-server-start.bat /kafka_2.13-3.6.1/bin/windows/zookeeper.properties
/kafka_2.13-3.6.1/bin/windows/zookeeper-server-start.bat /kafka_2.13-3.6.1/config/zookeeper.properties
# Window使用相对路径 系统找不到指定的路径
/bin/windows/zookeeper-server-start.bat /bin/config/zookeeper.properties
/kafka_2.13-3.6.1/bin/windows/zookeeper-server-start.bat /kafka_2.13-3.6.1/bin/config/zookeeper.properties
# 相对路径启动
call ./bin/windows/zookeeper-server-start.bat ./config/zookeeper.properties
call ./bin/windows/zookeeper-server-start.bat ./bin/windows/zookeeper.properties

ctrl + C停止，或者关闭窗口

3.启动kafka-server

# 当前路径下启动 将文件复制一份到启动命令处
kafka-server-start.sh config/server.properties
# 其他方式启动同上