自学内容网 自学内容网

ubuntu24.04部署单节点kafka_2.13-3.8.1

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 ]; then
        echo "Error: Failed to execute the last command."
        exit 1
    fi
}

# 函数:启动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; then
        echo "Error: Zookeeper failed to start."
        exit 1
    else
        echo "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; then
        echo "Error: Kafka broker failed to start."
        exit 1
    else
        echo "Kafka broker has been started with PID: $KAFKA_PID"
    fi
}

# 函数:停止Zookeeper和Kafka
stop_services() {
    if [ -f "$PID_FILE" ]; then
        while IFS= read -r line; do
            kill -9 "$line" 2>/dev/null && echo "Stopped service with PID: $line"
        done < "$PID_FILE"
        rm -f "$PID_FILE"
        echo "All services have been stopped."
    else
        echo "No PID file found, services may not be running."
    fi
}

# 主逻辑
if [ $# -eq 0 ]; then
    ACTION="start"
else
    ACTION="$1"
fi

case "$ACTION" in
    start)
        start_zookeeper
        start_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_command
        echo "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" in
    list)
        shift
        $KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --list
        ;;
    create)
        shift
        if [ $# -lt 6 ]; then
            echo "Error: Not enough arguments for creating a topic."
            show_help
            exit 1
        fi
        while [[ $# -gt 0 ]]; do
            case "$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_help
                    exit 1
                    ;;
            esac
        done
        $KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --create --replication-factor $REPLICATION_FACTOR --partitions $PARTITIONS --topic $TOPIC_NAME
        ;;
    produce)
        shift
        while [[ $# -gt 0 ]]; do
            case "$1" in
                --topic)
                    TOPIC_NAME="$2"
                    shift 2
                    ;;
                *)
                    echo "Unknown option: $1"
                    show_help
                    exit 1
                    ;;
            esac
        done
        if [ -z "$TOPIC_NAME" ]; then
            echo "Error: Topic name is required for producing messages."
            show_help
            exit 1
        fi
        $KAFKA_HOME/bin/kafka-console-producer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $TOPIC_NAME
        ;;
    consume)
        shift
        while [[ $# -gt 0 ]]; do
            case "$1" in
                --topic)
                    TOPIC_NAME="$2"
                    shift 2
                    ;;
                *)
                    echo "Unknown option: $1"
                    show_help
                    exit 1
                    ;;
            esac
        done
        if [ -z "$TOPIC_NAME" ]; then
            echo "Error: Topic name is required for consuming messages."
            show_help
            exit 1
        fi
        $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $TOPIC_NAME --from-beginning
        ;;
    *)
        show_help
        exit 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" in
    list)
        shift
        echo "正在列出所有主题..."
        $KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --list
        ;;
    create)
        shift
        if [ $# -lt 6 ]; then
            echo "错误: 创建主题时参数不足."
            show_help
            exit 1
        fi
        while [[ $# -gt 0 ]]; do
            case "$1" in
                --replication-factor)
                    REPLICATION_FACTOR="$2"
                    shift 2
                    ;;
                --partitions)
                    PARTITIONS="$2"
                    shift 2
                    ;;
                --topic)
                    TOPIC_NAME="$2"
                    shift 2
                    ;;
                *)
                    echo "未知选项: $1"
                    show_help
                    exit 1
                    ;;
            esac
        done
        echo "正在创建新主题: $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)
        shift
        while [[ $# -gt 0 ]]; do
            case "$1" in
                --topic)
                    TOPIC_NAME="$2"
                    shift 2
                    ;;
                *)
                    echo "未知选项: $1"
                    show_help
                    exit 1
                    ;;
            esac
        done
        if [ -z "$TOPIC_NAME" ]; then
            echo "错误: 发送消息时必须指定主题名称."
            show_help
            exit 1
        fi
        echo "正在启动控制台生产者,向主题 '$TOPIC_NAME' 发送消息..."
        $KAFKA_HOME/bin/kafka-console-producer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $TOPIC_NAME
        ;;
    consume)
        shift
        while [[ $# -gt 0 ]]; do
            case "$1" in
                --topic)
                    TOPIC_NAME="$2"
                    shift 2
                    ;;
                *)
                    echo "未知选项: $1"
                    show_help
                    exit 1
                    ;;
            esac
        done
        if [ -z "$TOPIC_NAME" ]; then
            echo "错误: 接收消息时必须指定主题名称."
            show_help
            exit 1
        fi
        echo "正在启动控制台消费者,从主题 '$TOPIC_NAME' 接收消息..."
        $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP_SERVER --topic $TOPIC_NAME --from-beginning
        ;;
    *)
        show_help
        exit 1
        ;;
esac

欢迎关注、您的支持与回复,是我继续开发更多技术文档分享的动力。欢迎扫描关注WX:夏目的杂货店


原文地址:https://blog.csdn.net/pblh123/article/details/144411763

免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!