Kafka：分布式消息系统的核心原理与安装部署-CSDN博客
自定义 Kafka 脚本 kf-use.sh 的解析与功能与应用示例-CSDN博客
Kafka 生产者全面解析：从基础原理到高级实践-CSDN博客
Kafka 生产者优化与数据处理经验-CSDN博客
Kafka 工作流程解析：从 Broker 工作原理、节点的服役、退役、副本的生成到数据存储与读写优化-CSDN博客
Kafka 消费者全面解析：原理、消费者 API 与Offset 位移-CSDN博客
Kafka 分区分配及再平衡策略深度解析与消费者事务和数据积压的简单介绍-CSDN博客
Kafka 数据倾斜：原因、影响与解决方案-CSDN博客
Kafka 核心要点解析_kafka mirrok-CSDN博客
Kafka 核心问题深度解析：全面理解分布式消息队列的关键要点_kafka队列日志-CSDN博客
目录
一、脚本功能概述
二、生产者性能测试
三、消费者性能测试
四、查看可用主题列表
五、脚本使用示例
六、脚本源码
七、总结
在大数据处理的领域中，Kafka 扮演着极为重要的角色，它作为一个分布式流处理平台，能够高效地处理大规模的实时数据。而 Kafka 提供了一系列的脚本工具，帮助我们更好地管理和测试 Kafka 集群。今天，我们就来深入探讨一个自定义的 Kafka 脚本 kf-use.sh，了解其功能与使用场景。
一、脚本功能概述
kf-use.sh 脚本主要提供了两个核心功能：生产者性能测试和消费者性能测试，同时还具备查看可用主题列表以及退出脚本等功能。通过这些功能，我们可以对 Kafka 集群的生产和消费能力进行评估，以便优化集群配置和数据处理流程。
二、生产者性能测试
- 参数输入
- 当选择生产者性能测试功能时，脚本首先会提示用户输入主题名称。如果输入的主题不存在，脚本会要求用户重新输入，确保测试的主题是有效的。
- 接着，用户需要输入要生产的记录数量，并且脚本会验证输入是否为整数，如果不是则提示重新输入。
- 最后，用户要输入每条记录的大小（单位为字节），同样会进行整数验证。
- 性能测试执行
- 一旦用户输入了正确的参数，脚本会调用
kafka-producer-perf-test.sh工具，并传入相应的参数，如主题名称、记录数量、记录大小等，同时设置吞吐量为 -1（表示不限制吞吐量），并指定 Kafka 集群的地址bigdata01:9092。这样就可以开始对生产者的性能进行测试，测试结果将反映出在给定条件下生产者向指定主题发送数据的效率。
- 一旦用户输入了正确的参数，脚本会调用
三、消费者性能测试
- 主题选择
- 在消费者性能测试功能中，首先会调用
kafka-topics.sh工具列出当前可用的主题列表，然后提示用户输入要测试的主题名称。如果输入的主题不存在，脚本会要求重新输入。
- 在消费者性能测试功能中，首先会调用
- 性能测试执行
- 当用户选择了存在的主题后，脚本会调用
kafka-consumer-perf-test.sh工具，传入 Kafka 集群地址bigdata01:9092、主题名称以及要消费的消息数量（这里固定为 100000 条），从而对消费者从指定主题消费数据的性能进行测试，测试结果可以帮助我们了解消费者处理数据的速度和效率。
- 当用户选择了存在的主题后，脚本会调用
四、查看可用主题列表
无论是在生产者还是消费者性能测试功能中，都提供了查看可用主题列表的选项。通过调用 kafka-topics.sh 工具并传入集群地址 bigdata01:9092，可以获取当前 Kafka 集群中所有的主题名称，方便用户了解集群中的数据主题情况，以便做出正确的测试选择。
五、脚本使用示例
假设我们要对一个名为 test_topic 的主题进行生产者性能测试，我们可以按照以下步骤操作：
- 运行
kf-use.sh脚本。 - 选择生产者性能测试功能（可能是对应的数字选项，如 1）。
- 输入主题名称
test_topic。 - 输入要生产的记录数量，例如 10000。
- 输入每条记录的大小，比如 100 字节。
- 脚本会自动执行生产者性能测试，并输出测试结果，包括发送的总字节数、每秒发送的记录数、每秒发送的字节数等信息。
同样，如果要进行消费者性能测试，例如对 test_topic 主题：
- 运行
kf-use.sh脚本并选择消费者性能测试功能（可能是数字 7 后再选择 1）。 - 查看可用主题列表，确认
test_topic存在后输入该主题名称。 - 脚本会执行消费者性能测试，并给出消费 100000 条消息的相关性能数据，如每秒消费的消息数等。
六、脚本源码
#!/bin/bash
while true; do
# 命令大全系统界面
echo "Kafka 命令大全系统："
echo "1. 主题操作（topics）"
echo "2. 生产者操作（producer）"
echo "3. 消费者操作（consumer）"
echo "4. 配置操作（configs）"
echo "5. 消费者组操作（consumer groups）"
echo "6. 生产者性能测试（producer perf test）"
echo "7. 消费者性能测试（consumer perf test）"
echo "0. 退出"
read -p "请输入功能选项数字：" choice
case $choice in
1)
# 主题操作菜单
while true; do
echo "主题操作功能："
echo "1. 查看所有主题"
echo "2. 创建主题"
echo "3. 查看某主题详细信息"
echo "4. 修改某主题分区数"
echo "5. 删除主题"
echo "0.返回命令大全系统界面"
read -p "请输入主题操作选项数字：" topic_choice
case $topic_choice in
1)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
;;
2)
# 创建主题
read -p "请输入要创建的主题名称：" topic_name
while true; do
read -p "请输入分区数（整数）：" partitions
if [[ $partitions =~ ^[0-9]+$ ]]; then
break
else
echo "分区数必须是整数，请重新输入。"
fi
done
while true; do
read -p "请输入副本数（整数，且不超过可用 broker 数量）：" replication_factor
if [[ $replication_factor =~ ^[0-9]+$ ]]; then
# 这里可添加检查副本数不超过可用 broker 数量的逻辑，暂时省略
break
else
echo "副本数必须是整数，请重新输入。"
fi
done
kafka-topics.sh --bootstrap-server bigdata01:9092 --create --partitions $partitions --replication-factor $replication_factor --topic $topic_name
;;
3)
# 查看某主题详细信息
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要查看详细信息的主题名称：" topic_to_describe
kafka-topics.sh --bootstrap-server bigdata01:9092 --describe --topic $topic_to_describe
;;
4)
# 修改某主题分区数
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要修改分区数的主题名称：" topic_to_alter
while true; do
read -p "请输入新的分区数（整数且大于当前分区数）：" new_partitions
if [[ $new_partitions =~ ^[0-9]+$ ]]; then
# 这里可添加检查新分区数是否大于当前分区数的逻辑，暂时省略
break
else
echo "新分区数必须是整数，请重新输入。"
fi
done
kafka-topics.sh --bootstrap-server bigdata01:9092 --alter --topic $topic_to_alter --partitions $new_partitions
;;
5)
# 删除主题
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要删除的主题名称：" topic_to_delete
kafka-topics.sh --bootstrap-server bigdata01:9092 --delete --topic $topic_to_delete
;;
0)
break
;;
*)
echo "无效的主题操作选择。"
;;
esac
done
;;
2)
# 生产者操作菜单
while true; do
echo "生产者操作功能："
echo "1. 发送消息到指定主题"
echo "0.返回命令大全系统界面"
read -p "请输入生产者操作选项数字：" producer_choice
case $producer_choice in
1)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要发送消息的主题名称：" topic_name
echo "开始发送消息到主题 $topic_name。输入'EXIT'退出发送。"
while true; do
read -p "请输入消息内容：" message
if [ "$message" = "EXIT" ]; then
break
fi
kafka-console-producer.sh --bootstrap-server bigdata01:9092 --topic $topic_name <<< "$message"
done
;;
0)
break
;;
*)
echo "无效的生产者操作选择。"
;;
esac
done
;;
3)
# 消费者操作菜单
while true; do
echo "消费者操作功能："
echo "1. 消费指定主题的消息"
echo "2. 从主题开头消费所有消息"
echo "0.返回命令大全系统界面"
read -p "请输入消费者操作选项数字：" consumer_choice
case $consumer_choice in
1)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要消费消息的主题名称：" topic_name
kafka-console-consumer.sh --bootstrap-server bigdata01:9092 --topic $topic_name
;;
2)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要从开头消费消息的主题名称：" topic_name
kafka-console-consumer.sh --bootstrap-server bigdata01:9092 --from-beginning --topic $topic_name
;;
0)
break
;;
*)
echo "无效的消费者操作选择。"
;;
esac
done
;;
4)
# 配置操作菜单
while true; do
echo "配置操作功能："
echo "1. 查看主题配置"
echo "2. 修改主题配置"
echo "0.返回命令大全系统界面"
read -p "请输入配置操作选项数字：" config_choice
case $config_choice in
1)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要查看配置的主题名称：" topic_name
kafka-configs.sh --bootstrap-server bigdata01:9092 --describe --entity-type topics --entity-name $topic_name
;;
2)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要修改配置的主题名称：" topic_name
read -p "请输入配置项名称：" config_name
read -p "请输入配置项值：" config_value
kafka-configs.sh --bootstrap-server bigdata01:9092 --alter --entity-type topics --entity-name $topic_name --add-config $config_name=$config_value
;;
0)
break
;;
*)
echo "无效的配置操作选择。"
;;
esac
done
;;
5)
# 消费者组操作菜单
while true; do
echo "消费者组操作功能："
echo "1. 查看消费者组列表"
echo "2. 查看消费者组详情"
echo "3. 重置消费者组偏移量"
echo "0.返回命令大全系统界面"
read -p "请输入消费者组操作选项数字：" consumer_group_choice
case $consumer_group_choice in
1)
kafka-consumer-groups.sh --bootstrap-server bigdata01:9092 --list
;;
2)
read -p "请输入要查看详情的消费者组名称：" group_name
kafka-consumer-groups.sh --bootstrap-server bigdata01:9092 --describe --group $group_name
;;
3)
read -p "请输入要重置偏移量的消费者组名称：" group_name
read -p "请输入要重置偏移量的主题名称：" topic_name
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
kafka-consumer-groups.sh --bootstrap-server bigdata01:9092 --reset-offsets --group $group_name --topic $topic_name --to-earliest
;;
0)
break
;;
*)
echo "无效的消费者组操作选择。"
;;
esac
done
;;
6)
# 生产者性能测试菜单
while true; do
echo "生产者性能测试功能："
echo "1. 执行生产者性能测试"
echo "2. 查看可用主题列表"
echo "0.返回命令大全系统界面"
read -p "请输入生产者性能测试选项数字：" producer_perf_choice
case $producer_perf_choice in
1)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要测试的主题名称：" topic_name
existing_topics=$(kafka-topics.sh --bootstrap-server bigdata01:9092 --list)
if echo "$existing_topics" | grep -q "$topic_name"; then
while true; do
read -p "请输入要发送的记录数量（整数）：" num_records
if [[ $num_records =~ ^[0-9]+$ ]]; then
break
else
echo "记录数量必须是整数，请重新输入。"
fi
done
while true; do
read -p "请输入每条记录的大小（整数，单位字节）：" record_size
if [[ $record_size =~ ^[0-9]+$ ]]; then
break
else
echo "记录大小必须是整数，请重新输入。"
fi
done
kafka-producer-perf-test.sh --topic $topic_name --num-records $num_records --record-size $record_size --throughput -1 --producer-props bootstrap.servers=bigdata01:9092
else
echo "主题 $topic_name 不存在，请重新输入。"
fi
;;
2)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
;;
0)
break
;;
*)
echo "无效的生产者性能测试选择。"
;;
esac
done
;;
7)
# 消费者性能测试菜单
while true; do
echo "消费者性能测试功能："
echo "1. 执行消费者性能测试"
echo "2. 查看可用主题列表"
echo "0.返回命令大全系统界面"
read -p "请输入消费者性能测试选项数字：" consumer_perf_choice
case $consumer_perf_choice in
1)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
echo "当前可用主题列表如下："
read -p "请输入要测试的主题名称：" topic_name
existing_topics=$(kafka-topics.sh --bootstrap-server bigdata01:9092 --list)
if echo "$existing_topics" | grep -q "$topic_name"; then
kafka-consumer-perf-test.sh --broker-list bigdata01:9092 --topic $topic_name --messages 100000
else
echo "主题 $topic_name 不存在，请重新输入。"
fi
;;
2)
kafka-topics.sh --bootstrap-server bigdata01:9092 --list
;;
0)
break
;;
*)
echo "无效的消费者性能测试选择。"
;;
esac
done
;;
0)
echo "退出脚本。"
break
;;
*)
echo "无效的选择。"
;;
esac
done七、总结
kf-use.sh 脚本为我们提供了一个便捷的方式来测试 Kafka 集群的生产者和消费者性能，同时方便地查看可用主题列表。通过合理使用这个脚本，我们可以更好地了解 Kafka 集群在数据生产和消费方面的能力，及时发现潜在的性能瓶颈并进行优化，从而提高整个大数据处理流程的效率。无论是对于 Kafka 初学者还是有一定经验的开发者，这个脚本都是一个非常实用的工具，可以帮助我们更好地管理和优化 Kafka 集群的运行。
在实际应用中，我们可以根据不同的业务需求和数据处理场景，灵活调整测试参数，深入分析测试结果，以确保 Kafka 集群能够稳定、高效地运行，满足日益增长的大数据处理需求。
