ubuntu24.04部署单节点kafka_2.13-3.8.1
下载地址推荐使用清华镜像源下载
https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/3.8.1/kafka_2.13-3.8.1.tgz
部署kafka部署
# 解压kafka压缩包
sudo tar -zxvf kafka_2.13-3.8.1.tgz -C /usr/local/# 改变权限及所有权
sudo chown -R hadoop:hadoopeco kafka_2.13-3.8.1/# 启动zookeeper
/usr/local/kafka_2.13-3.8.1$ ./bin/zookeeper-server-start.sh ./config/zookeeper.properties# 启动kafka
./bin/kafka-server-start.sh ./config/server.properties &# kafka操作
./bin/kafka-topics.sh --bootstrap-server lh007:9092 --list./bin/kafka-topics.sh --bootstrap-server lh007:9092 --create --replication-factor 1 --partitions 1 --topic test./bin/kafka-console-producer.sh --bootstrap-server lh007:9092 --topic test./bin/kafka-console-consumer.sh --bootstrap-server lh007:9092 --topic test
设置一键启动脚本
#!/bin/bash# 定义Kafka安装目录
KAFKA_HOME="/usr/local/kafka_2.13-3.8.1"
PID_FILE="kafka_pids.txt"# 函数：检查命令执行是否成功
check_command() {if [ $? -ne 0 ]; thenecho "Error: Failed to execute the last command."exit 1fi
}# 函数：启动Zookeeper，并等待其准备就绪
start_zookeeper() {echo "Starting Zookeeper..."$KAFKA_HOME/bin/zookeeper-server-start.sh $KAFKA_HOME/config/zookeeper.properties &ZOOKEEPER_PID=$!# 等待几秒钟以确保Zookeeper已经启动sleep 5# 检查Zookeeper进程是否存在if ! ps -p $ZOOKEEPER_PID > /dev/null; thenecho "Error: Zookeeper failed to start."exit 1elseecho "Zookeeper started with PID: $ZOOKEEPER_PID"fi
}# 函数：启动Kafka，并等待其准备就绪
start_kafka() {echo "Starting Kafka..."$KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties &KAFKA_PID=$!# 等待几秒钟以确保Kafka已经启动sleep 5# 检查Kafka进程是否存在if ! ps -p $KAFKA_PID > /dev/null; thenecho "Error: Kafka broker failed to start."exit 1elseecho "Kafka broker has been started with PID: $KAFKA_PID"fi
}# 函数：停止Zookeeper和Kafka
stop_services() {if [ -f "$PID_FILE" ]; thenwhile IFS= read -r line; dokill -9 "$line" 2>/dev/null && echo "Stopped service with PID: $line"done < "$PID_FILE"rm -f "$PID_FILE"echo "All services have been stopped."elseecho "No PID file found, services may not be running."fi
}# 主逻辑
if [ $# -eq 0 ]; thenACTION="start"
elseACTION="$1"
ficase "$ACTION" instart)start_zookeeperstart_kafka# 保存PIDs到文件，以便以后可以停止服务echo $ZOOKEEPER_PID > "$PID_FILE"echo $KAFKA_PID >> "$PID_FILE"echo "PIDs have been saved to $PID_FILE for stopping services later."check_commandecho "All services have been started successfully.";;stop)stop_services;;*)echo "Usage: $0 {start|stop}"exit 1;;
esac操作日志
hadoop@lh007:~/scripts$ ./kafka_control.sh start
Starting Zookeeper...
[2024-12-11 09:23:54,782] INFO Reading configuration from: /usr/local/kafka_2.13-3.8.1/config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,788] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,788] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,789] INFO observerMasterPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,789] INFO metricsProvider.className is org.apache.zookeeper.metrics.impl.DefaultMetricsProvider (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,792] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
[2024-12-11 09:23:54,792] INFO autopurge.purgeInterval set to 0 (org.apache.zookeeper.server.DatadirCleanupManager)
[2024-12-11 09:23:54,792] INFO Purge task is not scheduled. (org.apache.zookeeper.server.DatadirCleanupManager)
[2024-12-11 09:23:54,792] WARN Either no config or no quorum defined in config, running in standalone mode (org.apache.zookeeper.server.quorum.QuorumPeerMain)
[2024-12-11 09:23:54,794] INFO Log4j 1.2 jmx support not found; jmx disabled. (org.apache.zookeeper.jmx.ManagedUtil)
[2024-12-11 09:23:54,795] INFO Reading configuration from: /usr/local/kafka_2.13-3.8.1/config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,796] INFO clientPortAddress is 0.0.0.0:2181 (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,796] INFO secureClientPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,796] INFO observerMasterPort is not set (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,796] INFO metricsProvider.className is org.apache.zookeeper.metrics.impl.DefaultMetricsProvider (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2024-12-11 09:23:54,796] INFO Starting server (org.apache.zookeeper.server.ZooKeeperServerMain)
[2024-12-11 09:23:54,815] INFO ServerMetrics initialized with provider org.apache.zookeeper.metrics.impl.DefaultMetricsProvider@4590c9c3 (org.apache.zookeeper.server.ServerMetrics)
[2024-12-11 09:23:54,818] INFO ACL digest algorithm is: SHA1 (org.apache.zookeeper.server.auth.DigestAuthenticationProvider)
[2024-12-11 09:23:54,818] INFO zookeeper.DigestAuthenticationProvider.enabled = true (org.apache.zookeeper.server.auth.DigestAuthenticationProvider)
[2024-12-11 09:23:54,822] INFO zookeeper.snapshot.trust.empty : false (org.apache.zookeeper.server.persistence.FileTxnSnapLog)
[2024-12-11 09:23:54,835] INFO (org.apache.zookeeper.server.ZooKeeperServer)
[2024-12-11 09:23:54,836] INFO ______ _ (org.apache.zookeeper.server.ZooKeeperServer)
[2024-12-11 09:23:54,836] INFO |___ / | | (org.apache.zookeeper.server.ZooKeeperServer)
[2024-12-11 09:23:54,836] INFO / / ___ ___ | | __ ___ ___ _ __ ___ _ __ (org.apache.zookeeper.server.ZooKeeperServer)
[2024-12-11 09:23:54,836] INFO / / / _ \ / _ \ | |/ / / _ \ / _ \ | '_ \ / _ \ | '__| (org.apache.zookeeper.server.ZooKeeperServer)
[2024-12-11 09:23:54,836] INFO / /__ | (_) | | (_) | | < | __/ | __/ | |_) | | __/ | | (org.apache.zookeeper.server.ZooKeeperServer)
[2024-12-11 09:23:54,836] INFO /_____| \___/ \___/ |_|\_\ \___| \___| | .__/ \___| |_| (org.apache.zookeeper.server.ZooKeeperServer)epoch 0, of which 47 milliseconds was spent in the scheduler. (kafka.coordinator.group.GroupMetadataManager)
Kafka broker has been started with PID: 199769
PIDs have been saved to kafka_pids.txt for stopping services later.
All services have been started successfully.
hadoop@lh007:~/scripts$ jps
200310 Jps
199769 Kafka
199295 QuorumPeerMain
hadoop@lh007:~/scripts$ cat kafka_control.sh
#!/bin/bash操作脚本
hadoop@lh007:~/scripts$ cat my_kafka_options_en.sh
#!/bin/bash# 定义Kafka安装目录和其他常量
KAFKA_HOME="/usr/local/kafka_2.13-3.8.1"
BOOTSTRAP_SERVER="lh007:9092"# 函数：显示帮助信息
show_help() {echo "Usage: $0 {list|create|produce|consume} [options]"echo "Commands:"echo " list List all topics."echo " create Create a new topic with specified replication factor and partitions."echo " produce Start a console producer to send messages to the specified topic."echo " consume Start a console consumer to receive messages from the specified topic."echo "Options for 'create':"echo " --replication-factor <value>"echo " --partitions <value>"echo " --topic <name>"echo "Options for 'produce' and 'consume':"echo " --topic <name>"
}# 主逻辑
case "$1" inlist)shift$KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --list;;create)shiftif [ $# -lt 6 ]; thenecho "Error: Not enough arguments for creating a topic."show_helpexit 1fiwhile [[ $# -gt 0 ]]; docase "$1" in--replication-factor)REPLICATION_FACTOR="$2"shift 2;;--partitions)PARTITIONS="$2"shift 2;;--topic)TOPIC_NAME="$2"shift 2;;*)echo "Unknown option: $1"show_helpexit 1;;esacdone$KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --create --replication-factor $REPLICATION_FACTOR --partitions $PARTITIONS --topic $TOPIC_NAME;;produce)shiftwhile [[ $# -gt 0 ]]; docase "$1" in--topic)TOPIC_NAME="$2"shift 2;;*)echo "Unknown option: $1"show_helpexit 1;;esacdoneif [ -z "$TOPIC_NAME" ]; thenecho "Error: Topic name is required for producing messages."show_helpexit 1fi$KAFKA_HOME/bin/kafka-console-producer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $TOPIC_NAME;;consume)shiftwhile [[ $# -gt 0 ]]; docase "$1" in--topic)TOPIC_NAME="$2"shift 2;;*)echo "Unknown option: $1"show_helpexit 1;;esacdoneif [ -z "$TOPIC_NAME" ]; thenecho "Error: Topic name is required for consuming messages."show_helpexit 1fi$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $TOPIC_NAME --from-beginning;;*)show_helpexit 1;;
esac
中文版脚本
hadoop@lh007:~/scripts$ cat my_kafka_options_zn.sh
#!/bin/bash# 定义Kafka安装目录和其他常量
KAFKA_HOME="/usr/local/kafka_2.13-3.8.1"
BOOTSTRAP_SERVER="lh007:9092"# 函数：显示帮助信息（中文）
show_help() {echo "用法: $0 {list|create|produce|consume} [选项]"echo "命令:"echo " list 列出所有主题."echo " create 创建一个新主题，并指定复制因子和分区数."echo " produce 启动控制台生产者，向指定主题发送消息."echo " consume 启动控制台消费者，从指定主题接收消息."echo "创建主题的选项:"echo " --replication-factor <值> 设置复制因子"echo " --partitions <值> 设置分区数量"echo " --topic <名称> 指定主题名称"echo "启动生产和消费的选项:"echo " --topic <名称> 指定主题名称"
}# 主逻辑
case "$1" inlist)shiftecho "正在列出所有主题..."$KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --list;;create)shiftif [ $# -lt 6 ]; thenecho "错误: 创建主题时参数不足."show_helpexit 1fiwhile [[ $# -gt 0 ]]; docase "$1" in--replication-factor)REPLICATION_FACTOR="$2"shift 2;;--partitions)PARTITIONS="$2"shift 2;;--topic)TOPIC_NAME="$2"shift 2;;*)echo "未知选项: $1"show_helpexit 1;;esacdoneecho "正在创建新主题: $TOPIC_NAME, 复制因子: $REPLICATION_FACTOR, 分区数: $PARTITIONS..."$KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --create --replication-factor $REPLICATION_FACTOR --partitions $PARTITIONS --topic $TOPIC_NAME;;produce)shiftwhile [[ $# -gt 0 ]]; docase "$1" in--topic)TOPIC_NAME="$2"shift 2;;*)echo "未知选项: $1"show_helpexit 1;;esacdoneif [ -z "$TOPIC_NAME" ]; thenecho "错误: 发送消息时必须指定主题名称."show_helpexit 1fiecho "正在启动控制台生产者，向主题 '$TOPIC_NAME' 发送消息..."$KAFKA_HOME/bin/kafka-console-producer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $TOPIC_NAME;;consume)shiftwhile [[ $# -gt 0 ]]; docase "$1" in--topic)TOPIC_NAME="$2"shift 2;;*)echo "未知选项: $1"show_helpexit 1;;esacdoneif [ -z "$TOPIC_NAME" ]; thenecho "错误: 接收消息时必须指定主题名称."show_helpexit 1fiecho "正在启动控制台消费者，从主题 '$TOPIC_NAME' 接收消息..."$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $TOPIC_NAME --from-beginning;;*)show_helpexit 1;;
esac
欢迎关注、您的支持与回复，是我继续开发更多技术文档分享的动力。欢迎扫描关注WX：夏目的杂货店
