直接在机器上部署

下载

下载页面：https://kafka.apache.org/downloads
下载链接：https://dlcdn.apache.org/kafka/3.1.0/kafka_2.12-3.1.0.tgz

解压

tar -zxvf kafka_2.12-3.1.0.tgz -C /usr/local/ && cd /usr/local/kafka_2.12-3.1.0/config

配置server.properties

cd /usr/local/kafka_2.12-3.1.0/config
cp server.properties server-sample.properties
cd /usr/local/kafka_2.12-3.1.0/config && vim server.properties
mkdir -p /data/kafka-logs

  1. # Licensed to the Apache Software Foundation (ASF) under one or more
  2. # contributor license agreements.  See the NOTICE file distributed with
  3. # this work for additional information regarding copyright ownership.
  4. # The ASF licenses this file to You under the Apache License, Version 2.0
  5. # (the "License"); you may not use this file except in compliance with
  6. # the License.  You may obtain a copy of the License at
  7. #
  8. #    http://www.apache.org/licenses/LICENSE-2.0
  9. #
  10. # Unless required by applicable law or agreed to in writing, software
  11. # distributed under the License is distributed on an "AS IS" BASIS,
  12. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. # See the License for the specific language governing permissions and
  14. # limitations under the License.
  16. # see kafka.server.KafkaConfig for additional details and defaults
  18. ############################# Server Basics #############################
  20. # The id of the broker. This must be set to a unique integer for each broker.
  21. broker.id=1
  23. ############################# Socket Server Settings #############################
  25. # The address the socket server listens on. It will get the value returned from 
  26. # java.net.InetAddress.getCanonicalHostName() if not configured.
  27. #   FORMAT:
  28. #     listeners = listener_name://host_name:port
  29. #   EXAMPLE:
  30. #     listeners = PLAINTEXT://your.host.name:9092
  31. #listeners=PLAINTEXT://:9092
  32. listeners = PLAINTEXT://192.168.2.20:9092
  34. # Hostname and port the broker will advertise to producers and consumers. If not set, 
  35. # it uses the value for "listeners" if configured.  Otherwise, it will use the value
  36. # returned from java.net.InetAddress.getCanonicalHostName().
  37. #advertised.listeners=PLAINTEXT://your.host.name:9092
  39. # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
  40. #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
  42. # The number of threads that the server uses for receiving requests from the network and sending responses to the network
  43. num.network.threads=3
  45. # The number of threads that the server uses for processing requests, which may include disk I/O
  46. num.io.threads=8
  48. # The send buffer (SO_SNDBUF) used by the socket server
  49. socket.send.buffer.bytes=102400
  51. # The receive buffer (SO_RCVBUF) used by the socket server
  52. socket.receive.buffer.bytes=102400
  54. # The maximum size of a request that the socket server will accept (protection against OOM)
  55. socket.request.max.bytes=104857600
  58. ############################# Log Basics #############################
  60. # A comma separated list of directories under which to store log files
  61. # log.dirs=/tmp/kafka-logs
  62. log.dirs=/data/kafka-logs
  64. # The default number of log partitions per topic. More partitions allow greater
  65. # parallelism for consumption, but this will also result in more files across
  66. # the brokers.
  67. num.partitions=1
  69. # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
  70. # This value is recommended to be increased for installations with data dirs located in RAID array.
  71. num.recovery.threads.per.data.dir=1
  73. ############################# Internal Topic Settings  #############################
  74. # The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
  75. # For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
  76. offsets.topic.replication.factor=1
  77. transaction.state.log.replication.factor=1
  78. transaction.state.log.min.isr=1
  80. ############################# Log Flush Policy #############################
  82. # Messages are immediately written to the filesystem but by default we only fsync() to sync
  83. # the OS cache lazily. The following configurations control the flush of data to disk.
  84. # There are a few important trade-offs here:
  85. #    1. Durability: Unflushed data may be lost if you are not using replication.
  86. #    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
  87. #    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to excessive seeks.
  88. # The settings below allow one to configure the flush policy to flush data after a period of time or
  89. # every N messages (or both). This can be done globally and overridden on a per-topic basis.
  91. # The number of messages to accept before forcing a flush of data to disk
  92. #log.flush.interval.messages=10000
  94. # The maximum amount of time a message can sit in a log before we force a flush
  95. #log.flush.interval.ms=1000
  97. ############################# Log Retention Policy #############################
  99. # The following configurations control the disposal of log segments. The policy can
  100. # be set to delete segments after a period of time, or after a given size has accumulated.
  101. # A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
  102. # from the end of the log.
  104. # The minimum age of a log file to be eligible for deletion due to age
  105. log.retention.hours=168
  107. # A size-based retention policy for logs. Segments are pruned from the log unless the remaining
  108. # segments drop below log.retention.bytes. Functions independently of log.retention.hours.
  109. #log.retention.bytes=1073741824
  111. # The maximum size of a log segment file. When this size is reached a new log segment will be created.
  112. log.segment.bytes=1073741824
  114. # The interval at which log segments are checked to see if they can be deleted according
  115. # to the retention policies
  116. log.retention.check.interval.ms=300000
  118. ############################# Zookeeper #############################
  120. # Zookeeper connection string (see zookeeper docs for details).
  121. # This is a comma separated host:port pairs, each corresponding to a zk
  122. # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
  123. # You can also append an optional chroot string to the urls to specify the
  124. # root directory for all kafka znodes.
  125. # zookeeper.connect=localhost:2181
  126. zookeeper.connect=192.168.2.20:2181,192.168.2.21:2181,192.168.2.22:2181
  128. # Timeout in ms for connecting to zookeeper
  129. zookeeper.connection.timeout.ms=18000
  132. ############################# Group Coordinator Settings #############################
  134. # The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
  135. # The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
  136. # The default value for this is 3 seconds.
  137. # We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
  138. # However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
  139. group.initial.rebalance.delay.ms=0

其中：主要修改参数

  • broker.id : 当前机器服务器ID
  • log.dirs : 数据存放目录 log.dirs=/data/kafka-logs
  • zookeeper.connect : 机器Zookeeper地址 zookeeper.connect=192.168.2.20:2181,192.168.2.21:2181,192.168.2.22:2181
  • listeners修改为机器对应的ip：listeners = PLAINTEXT://192.168.2.20:9092

    配置jvm参数

    vim /usr/local/kafka_2.12-3.1.0/bin/kafka-server-start.sh
    export KAFKA_HEAP_OPTS=”-Xmx1G -Xms128M”

    分发安装包

  1. 拷贝整个目录到机器2、和机器3中
  2. scp -r /usr/local/kafka_2.12-3.1.0 root@192.168.2.21:/usr/local/kafka_2.12-3.1.0
  3. cd /usr/local/kafka_2.12-3.1.0/config && vim server.properties
  4. 修改对应的broker.id，另外两台机器分别设置id为2、3

启动集群

  1. 在master上建立群起脚本：startAllKafka.sh
  2. /usr/local/kafka_2.12-3.1.0/bin/kafka-server-start.sh -daemon /usr/local/kafka_2.12-3.1.0/config/server.properties
  3. 脚本内容如下 ```shell

    !/bin/bash

for host in “master” “node1” “node2” do echo “** $host **“ ssh $host “source /etc/profile; /usr/local/kafka_2.12-3.1.0/bin/kafka-server-start.sh -daemon /usr/local/kafka_2.12-3.1.0/config/server.properties” echo “*“ done

  2. 3. 启动集群
  3. ```shell
  4. [root@master ~]# ./startAllKafka.sh
  5. ********** master **********
  6. *******************************
  7. ********** node1 **********
  8. *******************************
  9. ********** node2 **********
  10. *******************************

注意：本项目配置的集群依赖外部Zookeeper集群，所以需要先启动Zookeeper机器

查看状态

  1. 在master上查看：

    1. [root@master ~]# jps -l
    2. 2341 org.apache.zookeeper.server.quorum.QuorumPeerMain
    3. 3847 sun.tools.jps.Jps
    4. 3517 kafka.Kafka

  2. 在node1上查看：

    1. [root@node1 kafka_2.12-3.1.0]# jps -l
    2. 2612 kafka.Kafka
    3. 2222 org.apache.zookeeper.server.quorum.QuorumPeerMain
    4. 2703 sun.tools.jps.Jps

  3. 在node2上查看：

    1. [root@node2 ~]# jps -l
    2. 2001 sun.tools.jps.Jps
    3. 1506 org.apache.zookeeper.server.quorum.QuorumPeerMain
    4. 1890 kafka.Kafka

    停止集群

  4. 在master上建立群停脚本：stopAllKafka.sh

  5. /usr/local/kafka_2.12-3.1.0/bin/kafka-server-stop.sh

  6. 脚本内容如下

    1. #! /bin/bash
    2. for host in "master" "node1" "node2"
    3. do
    4.      echo "********** $host **********"
    5.      ssh $host "source /etc/profile;/usr/local/kafka_2.12-3.1.0/bin/kafka-server-stop.sh"
    6.      echo "*******************************"
    7. done

  7. 停止集群 ```shell [root@master ~]# ./stopAllKafka.sh ** master **

** node1 **

** hadoop103 **

[root@master ~]# jps -l 2341 org.apache.zookeeper.server.quorum.QuorumPeerMain 4757 sun.tools.jps.Jps

  2. ---
  4. 至此，完成了Kafka集群的安装和启停。
  5. <a name="xFfZi"></a>
  6. ## 图形化工具
  7. 下载页面：[https://gitee.com/freakchicken/kafka-ui-lite/](https://gitee.com/freakchicken/kafka-ui-lite/)<br />下载链接：[https://gitee.com/freakchicken/kafka-ui-lite/repository/archive/v1.2.11?format=zip](https://gitee.com/freakchicken/kafka-ui-lite/repository/archive/v1.2.11?format=zip)<br />java -jar kafka-ui-lite.jar<br />启动后浏览器访问 [http://ip:8889](https://gitee.com/link?target=http%3A%2F%2Fip%3A8889)
  9. docker安装<br />docker run -d -p 8889:8889 freakchicken/kafka-ui-lite
  12. <a name="I2Opc"></a>
  13. # 使用docker部署
  14. ```shell
  16. docker run --network host \
  17. -v /data/zookeeper_data/data:/data \
  18. -v /data/zookeeper_data/conf:/conf \
  19. --name zookeeper_1 \
  20. -d bitnami/zookeeper:3.6.3
  22. docker run -it --rm \
  23. --network app-tier \
  24. -e KAFKA_CFG_ZOOKEEPER_CONNECT=192.168.2.20:31808 \
  25. bitnami/kafka:2.8.1 kafka-topics.sh --list  --bootstrap-server localhost:9092

使用kubesphere部署

dockerhub

redis:6.2.6

资源分配

image.png

端口配置

image.png

启动命令

redis-server
/etc/redis/redis.conf
image.png

环境变量

KAFKA_CFG_ZOOKEEPER_CONNECT
ALLOW_PLAINTEXT_LISTENER
KAFKA_ADVERTISED_LISTENERS PLAINTEXT://192.168.2.20:31374
KAFKA_LISTENERS

image.png

同步主机时区

image.png

存储卷配置

/data
image.png

配置文件挂载（需要提前创建configMap）

/etc/redis/redis.conf
image.png

配置文件子路径挂载

（* 默认使用路径挂载会覆盖容器内所有配置，所以如果只需要覆盖个别配置可以使用子路径）

/home/nacos/config/application.properties

图片.png

/home/nacos/config/cluster.conf图片.png

参考

docker安装kafka：https://blog.csdn.net/qq_22041375/article/details/106180415