- 官网下载地址
https://kafka.apache.org/downloads
- 1、上传解压
tar -zxvf kafka_2.11-1.0.0.tgz -C ../
mv kafka_2.11-1.0.0 kafka-1.0.0
- 2、修改环境变量
配置环境变量
vim /etc/profile
export KAFKA_HOME=/usr/local/soft/kafka-1.0.0
export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile
- 3、修改配置文件
vim config/server.properties
broker.id=0 每一个节点broker.id 要不一样
zookeeper.connect=master:2181,node1:2181,node2:2181/kafka
log.dirs=/usr/local/soft/kafka-1.0.0/data 数据存放的位置
- 4、将kafka文件同步到node1,node2
scp -r kafka-1.0.0/ node1:
pwd
scp -r kafka-1.0.0/ node2:pwd
修改node1、node2中的/etc/profile，增加Kafka环境变量
export KAFKA_HOME=/usr/local/soft/kafka-1.0.0export PATH=$PATH:$KAFKA_HOME/bin
source /etc/profile
- 5、修改node1和node2中的broker.id
vim config/server.properties
node1
broker.id=1
node2
broker.id=2
- 6、启动kafka
需要启动zookeeper, kafka使用zk保存元数据
zkServer.sh start
启动kafka，每个节点中都要启动（去中心化的架构）
-daemon后台启动
kafka-server-start.sh -daemon /usr/local/soft/kafka-1.0.0/config/server.properties
- 7、测试是否成功
生产者
消费者
kafka的使用
- 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 hhx
- 3、获取所有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、创建控制台消费者
--from-beginning 从头消费，， 如果不在执行消费的新的数据
kafka-console-consumer.sh --bootstrap-server master:9092,node1:9092,node3:9092 --from-beginning --topic bigdata
- 6、kafka数据保存的方式
# 1、保存的文件
/usr/local/soft/kafka_2.11-1.0.0/data# 2，每一个分区每一个副本对应一个目录# 3、每一个分区目录中可以有多个文件， 文件时滚动生成的
00000000000000000000.log
00000000000000000001.log
00000000000000000002.log# 4、滚动生成文件的策略
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000# 5、文件删除的策略，默认时7天，以文件为单位删除
log.retention.hours=168
