kafka基本命令
!!!!!先说一个容易失败的地方,如果是Windows必须要把下载下来的kafka改一下名字,去掉后面的版本号,不然后面运行命令的时候,他会出现命令过长导致不能运行,非常重要
配置kafka模式
生成集群id
bin\windows\kafka-storage.bat random-uuid手动赋值
set KAFKA_CLUSTER_ID=这里粘贴你拿到的uuid字符串格式化存储目录,自己替换一下集群id就可以f-z1egYyST0pH5ivw7xulQ替换掉
bin\windows\kafka-storage.bat format -t f-z1egYyST0pH5ivw7xulQ -c config\kraft\server.properties启动kafka
bin\windows\kafka-server-start.bat config\kraft\server.properties
bin\windows\kafka-server-start.bat config\kraft\server.properties配置脚本:Windows用ps1(一开始想用bat但是bat不好用该用ps1,bat文件你就不用管,复制下面代码粘进去)
# Kafka KRaft mode startup script for Windows PowerShell # Run this script from the Kafka root directory. $ErrorActionPreference = "Stop" $CurrentRoot = [System.IO.Path]::GetPathRoot((Get-Location).Path) $LogDir = Join-Path $CurrentRoot "tmp\kraft-combined-logs" $ConfigFile = "config\kraft\server.properties" $KafkaStorage = ".\bin\windows\kafka-storage.bat" $KafkaServerStart = ".\bin\windows\kafka-server-start.bat" Write-Host "=== Kafka KRaft Mode Startup Script ===" Write-Host "Current time: $(Get-Date)" Write-Host "Working directory: $(Get-Location)" Write-Host "" function Cleanup-Logs { if (Test-Path -LiteralPath $LogDir) { Write-Host "Existing log directory found. Cleaning..." Remove-Item -LiteralPath $LogDir -Recurse -Force Write-Host "Log directory cleaned: $LogDir" } else { Write-Host "Log directory does not exist. No cleanup needed." } } function Format-Storage { Write-Host "" Write-Host "=== Step 1: Generate cluster ID ===" $KafkaClusterId = (& $KafkaStorage random-uuid).Trim() Write-Host "Generated cluster ID: $KafkaClusterId" Write-Host "" Write-Host "=== Step 2: Format storage directory ===" & $KafkaStorage format -t $KafkaClusterId -c $ConfigFile if ($LASTEXITCODE -eq 0) { Write-Host "Storage formatted successfully." return $true } Write-Host "Storage format failed. This may be caused by a cluster ID conflict." return $false } function Start-Kafka { Write-Host "" Write-Host "=== Step 3: Start Kafka server ===" Write-Host "Config file: $ConfigFile" Write-Host "Starting Kafka..." Write-Host "Tip: Press Ctrl+C to stop the server." Write-Host "" & $KafkaServerStart $ConfigFile } function Main { if (-not (Format-Storage)) { Write-Host "" Write-Host "Cluster ID conflict detected. Cleaning logs and retrying..." Cleanup-Logs Write-Host "" Write-Host "=== Retry storage format ===" if (-not (Format-Storage)) { Write-Host "Still failed after retry. Please check your Kafka config." exit 1 } } Start-Kafka } if (-not (Test-Path -LiteralPath $ConfigFile)) { Write-Host "Config file does not exist: $ConfigFile" exit 1 } if (-not (Test-Path -LiteralPath $KafkaStorage)) { Write-Host "kafka-storage.bat does not exist. Please run this script from the Kafka root directory." exit 1 } if (-not (Test-Path -LiteralPath $KafkaServerStart)) { Write-Host "kafka-server-start.bat does not exist. Please check your Kafka installation." exit 1 } Main注意:运用脚本启动的时候就不要用前面的命令启动了,前面命令启动的也要关掉,否则第三步会失败
验证安装:
bin\windows\kafka-topics.bat --create --topic quickstart-events --bootstrap-server localhost:9092创建主题:
bin\windows\kafka-topics.bat --create --topic quickstart-events --bootstrap-server localhost:9092查看全部主题列表
bin\windows\kafka-topics.bat --list --bootstrap-server localhost:9092查看指定主题详细
bin\windows\kafka-topics.bat --describe --topic quickstart-events --bootstrap-server localhost:9092测试生产者和消费者(中文传输会乱码)
bin\windows\kafka-console-producer.bat --topic quickstart-events --bootstrap-server localhost:9092bin\windows\kafka-console-consumer.bat --topic quickstart-events --from-beginning --bootstrap-server localhost:9092创建聪明需要的主题
创建文件处理主题
bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic file-processing创建向量化主题
bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic vectorization创建死信队列主题
bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic file-processing-dlt关闭的时候最好ctrl+c关闭,异常关闭下次启动会异常
如果异常关闭了:
删掉,重新来一次就重新生成了tmp里面的文件