部署
官网地址：https://kafka.apache.org/
官网下载地址：https://kafka.apache.org/downloads
注：Kafka是用Scala语言开发的，运行在JVM上，因此需要jdk做为基础环境，使用jdk1.8以上的版本。
找到kafka_2.13-2.8.1.tgz包进行下载。详细的安装步骤如下：
-
创建目录上传jdk文件
mkdir /soft cd /soft -
解压kafka压缩文件
tar -zxvf kafka_2.13-2.8.1.tgz -
提前创建配置文件中所有的文件夹
mkdir -p /opt/software/kafka # zookeeper数据目录 mkdir -p /soft/kafka_2.13-2.8.1/zookeeper # kafka日志 mkdir -p /soft/kafka_2.13-2.8.1/kafka/log # zookeeper日志 mkdir -p /soft/kafka_2.13-2.8.1/zookeeper/log -
修改zookeeper的配置文件
zookeeper.properties# 切换目录 cd /soft/kafka_2.13-2.8.1/config # 编辑 zookeeper配置文件 vim zookeeper.properties # zookeeper数据目录 dataDir=/soft/kafka_2.13-2.8.1/zookeeper # zookeeper日志目录 dataLogDir=/soft/kafka_2.13-2.8.1/zookeeper/log # 更改zookeeper端口 clientPort=2181 maxClientCnxns=100 tickTimes=2000 initLimit=10 syncLimit=5 -
修改kafka的配置文件
server.properties# 切换目录 cd /soft/kafka_2.13-2.8.1/config # 编辑 kafka配置文件 vim server.properties ###### server.properties开始 ###### # 设置ip端口 (阿里云服务器，使用内网ip地址) listeners=PLAINTEXT://172.17.123.108:9092 #日志存放路径，上面创建的目录 log.dirs=/soft/kafka_2.13-2.8.1/kafka/log #zookeeper地址和端口，单机配置部署，localhost:2181 zookeeper.connect=172.17.123.108:2181 ###### server.properties结束 ###### # 查看服务器网卡ip 作为kafka 和 zookeeper 的ip ip addr -
启动kafka
# 切换目录 cd /soft/kafka_2.13-2.8.1/bin # 优先启动 zookeeper ./zookeeper-server-start.sh /soft/kafka_2.13-2.8.1/config/zookeeper.properties & # 再启动 kafka ./kafka-server-start.sh /soft/kafka_2.13-2.8.1/config/server.properties &
基础指令
以下主要介绍topic的创建、插件，生产者发送消息，消费者消费消息的基础脚本指令的使用
# 查看topic数据
./kafka-topics.sh --bootstrap-server 8.140.224.210:9092 --list
# 创建topic
./kafka-topics.sh --create --topic <your_topic_name> --bootstrap-server localhost:9092 --partitions <num_partitions> --replication-factor <replication_factor>
--topic <your_topic_name>：指定你要创建的 Topic 名称。
--bootstrap-server localhost:9092：Kafka 服务的地址。
--partitions <num_partitions>：为该 Topic 指定的分区数量。
--replication-factor <replication_factor>：副本因子，单机环境下通常设置为 1。
./kafka-topics.sh --create --topic data0912 --bootstrap-server 8.140.224.210:9092 --partitions 1 --replication-factor 1
# 生产消息
./kafka-console-producer.sh --topic data0912 --bootstrap-server 8.140.224.210:9092
# 消费消息
./kafka-console-consumer.sh --topic data0912 --bootstrap-server 8.140.224.210:9092 --from-beginning
服务运维
-
服务器磁盘爆满，导致kafka服务崩溃
当别人通过kafka给我推送十几亿文本数据时，我正常批量读取数据，且多线程的处理数据，也验证了消费数据是极快的，不会造成kafka数据的挤压。但是还是导致了kafka服务的崩溃，排查后发现云服务器100GB的磁盘已经爆满，便在排查时发现kakfa默认保存近七天的数据，而我一天的数据就超过了100GB。具体的解决方案如下：
# 1.查看kafka的配置文件 server.properties 中log文件的存放位置 log.dirs=/soft/kafka_2.13-2.8.1/kafka/log cd /soft/kafka_2.13-2.8.1/config cat server.properties # 2.切换到log文件下查看，那个topic目录下的磁盘空间占用过大 cd /soft/kafka_2.13-2.8.1/kafka/log # 查看log文件下占用磁盘严重的目录 du -sh * # 切换到 对应的目录 cd cd data0912-0 #3.删除较早的***.log文件 # 目录下存在众多***.index,***.log文件；***.index中存放的消息物理位置的偏移量；***.log中存放的是消息的实际内容 # log文件默认存放时间为 7天；7天内所有数据都进行存储，就会导致磁盘爆破的问题 rm -rf *.log #4.更改配置文件 cd /soft/kafka_2.13-2.8.1/config vim server.properties # 更改以下配置 # segment 文件保留时间（默认168小时 7天） #log.retention.hours=168 # segment 文件的保留时间改为 5小时 log.retention.hours=5 # 日志文件中 每个segment 文件默认大小 为1g log.segment.bytes=1073741824 # 多久时间扫描一次 日志文件的大小 （单位：毫秒） 五分钟 log.retention.check.interval.ms=300000 # kakfa 从0.10.0版本之后不再需要通过以下配置来启动日志清理策略；直接按照 留存时间、文件大小，进行清理 # 日志清理配置 开始 # 以segment为基本单位进行定期清理，当前正在使用的segment不会被清理。 # 日志清理是否打开 log.cleaner.enable=ture # 多线程清理，提高数据清理速度 log.cleaner.threads=2 # 采用日志删除策略进行删除 log.cleanup.policy=delete # 日志清理配置 结束 #5.重启kafka
