集群使用Kafka

创建Kafka启动脚本

ZooKeeper为Kafka提供了集群管理、分布式协调、配置管理和高可用性等关键服务，Kafka启动之前需要先zkServer.sh start。

Kafka 2.8.0版本开始引入了自身的内部元数据管理机制（KRaft模式），以逐步替代ZooKeeper。

kafka/bin目录下，例如 /usr/local/soft/kafka-1.0.0/bin

vim start-kafka.sh

获得执行权限

chmod +x  start-kafka.sh

#!/usr/bin/env bash

# 检查Kafka启动脚本是否存在
if [[ ! -x "/usr/local/soft/kafka-1.0.0/bin/kafka-server-start.sh" ]]; then
  echo "错误：找不到Kafka启动脚本或没有执行权限"
  exit 1
fi

# 检查Kafka配置文件是否存在
if [[ ! -f "/usr/local/soft/kafka-1.0.0/config/server.properties" ]]; then
  echo "错误：找不到Kafka配置文件"
  exit 1
fi


# 使用绝对路径启动Kafka，确保脚本能够正确执行
/usr/local/soft/kafka-1.0.0/bin/kafka-server-start.sh -daemon /usr/local/soft/kafka-1.0.0/config/server.properties

# 检查Kafka是否成功启动
# 显示等待信息，并每秒打印一个“.”
echo -n "正在启动Kafka,请等待5秒"
for i in {1..5}; do
  echo -n "."
  sleep 1
done
echo

# 使用ps命令检查Kafka进程是否启动
PIDS=$(ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}')
if [[ -n "$PIDS" ]]; then
  echo "Kafka启动成功，进程ID: $PIDS"
else
  echo "Kafka启动失败，请检查日志文件 $KAFKA_LOG_FILE 以获取更多信息"
  exit 1
fi

停止脚本

同样在bin目录下，

vim stop-kafka.sh

chmod +x  stop-kafka.sh 

#!/usr/bin/env bash

# 检查Kafka停止脚本是否存在
if [[ ! -x "/usr/local/soft/kafka-1.0.0/bin/kafka-server-stop.sh" ]]; then
  echo "错误：找不到Kafka停止脚本或没有执行权限"
  exit 1
fi

# 调用Kafka停止脚本
/usr/local/soft/kafka-1.0.0/bin/kafka-server-stop.sh

# 显示等待信息，并每秒打印一个“.”
echo -n "正在停止Kafka,请等待5秒"
for i in {1..5}; do
  echo -n "."
  sleep 1
done
echo

# 检查Kafka是否成功停止
if pgrep -f kafka.Kafka > /dev/null; then
  echo "错误：Kafka未能成功停止"
  exit 1
else
  echo "Kafka已成功停止"
fi

1、创建topic

在生产和消费数据时，如果topic不存在会自动创建一个分区为1，副本为1的topic

--replication-factor  ---每一个分区的副本数量, 同一个分区的副本不能放在同一个节点，副本的数量不能大于kafak集群节点的数量
--partition   --分区数，  根据数据量设置
--zookeeper zk的地址，将topic的元数据保存在zookeeper中

kafka-topics.sh --create --zookeeper master:2181,node1:2181,node2:2181/kafka --replication-factor 2 --partitions 3 --topic bigdata

2、查看topic描述信息

kafka-topics.sh --describe  --zookeeper master:2181,node1:2181,node2:2181/kafka --topic shujia

3、获取所有topic

__consumer_offsetsL kafka用于保存消费便宜量的topic

kafka-topics.sh --list  --zookeeper  master:2181,node1:2181,node2:2181/kafka

4、创建控制台生产者

kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic bigdata

5、创建控制台消费者

x  --from-beginning  从头消费，， 如果不在执行消费的新的数据

kafka-console-consumer.sh --bootstrap-server  master:9092,node1:9092,node3:9092 --from-beginning --topic bigdata