环境：Centos7 软件：kafka_2.11-2.4.1.tgz 安装步骤： 下载安装 配置 部署 验证

安装kafka前，需要安装zookeeper为其提供分布式服务。可参考以往文章Zookeeper-集群部署。
集群规划：

hadoop001 hadoop002 hadoop003
zookeeper zookeeper zookeeper
kafka kafka kafka

下载安装

官网-kafka_2.11-2.4.1.tgz。

  1. tar -zxvf kafka_2.11-2.4.1.tgz 
  2. mv kafka_2.11-2.4.1 kafka

配置

kafka目录下创建logs文件夹，用于保存kafka运行过程中产生的日志文件。

kafka/config/server.properties

broker.id 必须唯一，在hadoop002和hadoop003中分别修改即可

  1. mkdir kafka/logs
  2. cd kafka/config
  3. vi server.properties
  5. ############################# Server Basics #############################
  7. broker.id=0
  9. # 监听地址
  10. listeners=PLAINTEXT://hadoop001:9092
  11. ############################# Socket Server Settings #############################
  13. # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
  14. #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
  15. delete.topic.enable=true
  16. # The number of threads that the server uses for receiving requests from the network and sending responses to the network
  17. num.network.threads=3
  19. # The number of threads that the server uses for processing requests, which may include disk I/O
  20. num.io.threads=8
  22. # The send buffer (SO_SNDBUF) used by the socket server
  23. socket.send.buffer.bytes=102400
  25. # The receive buffer (SO_RCVBUF) used by the socket server
  26. socket.receive.buffer.bytes=102400
  28. # The maximum size of a request that the socket server will accept (protection against OOM)
  29. socket.request.max.bytes=104857600
  31. ############################# Log Basics #############################
  33. # A comma separated list of directories under which to store log files
  34. log.dirs=/root/kafka/logs
  36. # The default number of log partitions per topic. More partitions allow greater
  37. # parallelism for consumption, but this will also result in more files across
  38. # the brokers.
  39. num.partitions=1
  41. # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
  42. # This value is recommended to be increased for installations with data dirs located in RAID array.
  43. num.recovery.threads.per.data.dir=1
  45. ############################# Zookeeper #############################
  47. # Zookeeper connection string (see zookeeper docs for details).
  48. # This is a comma separated host:port pairs, each corresponding to a zk
  49. # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
  50. # You can also append an optional chroot string to the urls to specify the
  51. # root directory for all kafka znodes.
  52. zookeeper.connect=hadoop001:2181,hadoop002:2181,hadoop003:2181
  54. # Timeout in ms for connecting to zookeeper
  55. zookeeper.connection.timeout.ms=6000

注：如出现
image.png
按提示修改broker.id即可。

配置环境变量

  1. export KAFKA_HOME=/root/kafka
  2. export Path=$PATH:$KAFKA_HOME/bin

部署

分发程序

  1. scp -r /root/kafka/ hadoop002:/root/
  2. scp -r /root/kafka/ hadoop003:/root/

启动与停止

启动： kafka-server-start.sh server.properties

  1. /root/kafka/bin/kafka-server-start.sh /root/kafka/config/server.properties &

启动成功：
image.png
image.png
image.png
停止：kafka-server-stop.sh stop

  1. /root/kafka/bin/kafka-server-stop.sh stop

image.png

测试

  1. kafka/bin/kafka-topics.sh --create --bootstrap-server hadoop001:9092 \
  2.                     --replication-factor 3 \
  3.                     --partitions 1 --topic my-replicated-topic
  1. kafka/bin/kafka-topics.sh --describe --bootstrap-server hadoop001:9092 --topic my-replicated-topic

结果如图：
image.png