部署安装单节点

  • 节点IP：192.168.0.10
  1. 从apache或者国内镜像源下载安装包，这里用ustc示例：
  1. wget http://mirrors.ustc.edu.cn/apache/kafka/2.7.0/kafka_2.13-2.7.0.tgz
  1. 解压，假设解压到 /home/zhangsan/apps 目录
  1. tar xf kafka_2.13-2.7.0.tgz -C /home/zhangsan/apps
  1. 启动包里自带的zookeeper
  1. cd /home/zhangsan/apps/kafka_2.13-2.7.0
  2. # 如有需要，可以修改配置文件中zookeeper数据的存储目录
  3. ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties
  1. 修改kafka配置文件
  1. # 配置监听的IP和端口
  2. # 如有需要，可以修改数据存储目录和过期时间
  3. sed -i "s@#listeners=PLAINTEXT://:9092@listeners=PLAINTEXT://192.168.0.10:9092@g" ./config/server.properties
  1. 启动kafka
  1. ./bin/kafka-server-start.sh -daemon ./config/server.properties
  1. 验证。开两个终端，一个启动消费者，一个启动生产者：
  1. # 在一个终端启动消费者
  2. ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.10:9092 --topic test --from-beginning
  4. # 在另一个终端启动生产者
  5. ./bin/kafka-console-producer.sh --broker-list 192.168.0.10:9092 --topic test
  7. # 在生产者终端随便输入，如果在消费者终端看到相应输出，说明部署成功。
  • 停止kafka
  1. ./bin/kafka-server-stop.sh
  • 停止自带的zookeeper
  1. ./bin/zookeeper-server-stop.sh

部署安装kafka集群

环境

zookepper集群节点IP和端口：

  • 192.168.0.111:2181
  • 192.168.0.112:2181
  • 192.168.0.113:2181

kafka集群节点IP和端口：

  • 192.168.0.11:9092
  • 192.168.0.12:9092
  • 192.168.0.11:9092

kafka版本：2.13-2.7.0

步骤

  1. 下载解压步骤同单节点，此处略过
  2. 修改kafka配置文件，配置zookeeper地址及其它参数，以192.168.0.11为例：
  1. # 每一个Broker在集群中的唯一标识，必须为正数
  2. broker.id=0
  4. # kafka监听地址与端口
  5. listeners=PLAINTEXT://192.168.0.11:9092
  7. # broker处理消息的最大线程数（默认为3），可以设置为CPU核数+1
  8. num.network.threads=3
  10. # broker处理磁盘IO的线程数，可以设置为CPN核数*2
  11. num.io.threads=8
  12. socket.send.buffer.bytes=102400
  13. socket.receive.buffer.bytes=102400
  14. socket.request.max.bytes=104857600
  16. # kafka保存数据的位置
  17. log.dirs=/home/zhangsan/data/kafka
  19. # 设置新创建的分区有多少个分区，可以根据消费者实际情况配置，配置过小会影响消费性能
  20. num.partitions=6
  22. num.recovery.threads.per.data.dir=1
  24. offsets.topic.replication.factor=1
  25. transaction.state.log.replication.factor=1
  26. transaction.state.log.min.isr=1
  28. # 数据保存72小时，默认168小时
  29. log.retention.hours=72
  31. # 分区中每个段数据文件的大小为1GB
  32. log.segment.bytes=1073741824
  33. log.retention.check.interval.ms=300000
  35. # zookeeper所在地址
  36. zookeeper.connect=192.168.0.111:2181,192.168.0.112:2181,192.168.0.113:2181,
  37. zookeeper.connection.timeout.ms=18000
  39. group.initial.rebalance.delay.ms=0
  40. # 自动创建主题
  41. auto.create.topics.enable=true
  42. # 提供删除主题的功能
  43. delete.topic.enable=true
  1. 启动三个节点上的kafka，同单节点的启动步骤。
  2. 验证
  1. # 192.168.0.11上启动生产者
  2. ./bin/kafka-console-producer.sh --broker-list 192.168.0.11:9092,192.168.0.12:9092,192.168.0.13:9092 --topic testtopic
  4. # 192.168.0.12和192.168.0.13上启动消费者
  5. ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.11:9092,192.168.0.12:9092,192.168.0.13:9092 --topic testtopic

其它命令

  • 显示主题列表
  1. ./bin/kafka-topics.sh --zookeeper 192.168.0.111:2181,192.168.0.112:2181,192.168.0.113:2181 --list
  • 创建主题
  1. ./bin/kafka-topics.sh --create --zookeeper 192.168.0.111:2181,192.168.0.112:2181,192.168.0.113:2181 --replication-factor 1 --partitions 3 --topic testtopic
  • 查看主题状态
  1. ./bin/kafka-topics.sh --describe --zookeeper 192.168.0.111:2181,192.168.0.112:2181,192.168.0.113:2181 --topic testtopic
  • 删除主题
  1. ./bin/kafka-topics.sh --zookeeper 192.168.0.111:2181,192.168.0.112:2181,192.168.0.113:2181 --delete --topic testtopic