0
0

【中间件】通过 docker-compose 快速部署 Kafka 保姆级教程

文章目录

一、概述

Kafka是由Apache基金会开发的分布式流处理平台，采用发布-订阅模式，支持高吞吐量、低延迟的数据传输。主要用于处理实时数据管道、数据存储和数据分析等大数据应用场景。Kafka采用高效的数据压缩算法，可以在集群中存储大量的数据，并通过分区机制来实现数据的高可靠性和可扩展性。Kafka常用于以下场景：

  • 数据管道：在数据采集和分发过程中构建可扩展的流式数据管道，用于实时数据处理和分析。例如，数据收集、日志聚合、网络追踪、用户活动跟踪等。
  • 数据存储：将Kafka作为持久化存储来存储大量的数据，以便用于后续的批量处理和离线分析，例如数据挖掘、机器学习等应用场景。
  • 实时流处理：通过将Kafka与追求低延迟的流式处理平台，例如Apache Storm、Apache Samza和Apache Flink等相结合，可以实现实时数据处理和分析。这是许多实时数据分析和日志处理需求的主要场景。
  • 系统日志跟踪：通过Kafka将来自不同系统的日志数据统一收集和存储，便于进行统一的日志分析和事件跟踪，在软件开发过程中可以快速定位和解决问题。

总之，Kafka是高性能、可靠、可扩展的分布式流处理平台，可用于实时数据管道、数据存储、实时流处理和日志跟踪等多个领域。它已被广泛应用于各种大数据场景，并成为了大数据架构中的一个重要组成部分。

在这里插入图片描述
这里只是讲解kafka容器快速部署，用于测试和学习作用，生成不建议使用容器部署，想了解更多的kafka知识点可参考我这篇文章：Kafka原理介绍+安装+基本操作

二、前期准备

1）部署 docker

  1. # 安装yum-config-manager配置工具
  2. yum -y install yum-utils
  4. # 建议使用阿里云yum源：（推荐）#yum-config-manager --add-repo https://download.docker.com/linux/centos/docker-ce.repo
  5. yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo
  7. # 安装docker-ce版本
  8. yum install -y docker-ce
  9. # 启动并开机启动
  10. systemctl enable --now docker
  11. docker --version

2）部署 docker-compose

  1. curl -SL https://github.com/docker/compose/releases/download/v2.16.0/docker-compose-linux-x86_64 -o /usr/local/bin/docker-compose
  3. chmod +x /usr/local/bin/docker-compose
  4. docker-compose --version

三、创建网络

  1. # 创建，注意不能使用hadoop_network，要不然启动hs2服务的时候会有问题！！！
  2. docker network create hadoop-network
  4. # 查看
  5. docker network ls

四、安装 Zookeeper

Zookeeper在Kafka中扮演重要的角色，主要用于管理Kafka集群的元数据和实现Kafka集群的协调和管理。在Kafka集群中，Zookeeper主要有以下作用：

  • 配置管理：Kafka集群的配置信息存储在ZK节点中，包括Kafka Broker的配置信息、Topic的分区信息、消费者和生产者的相关配置等。Kafka可以通过ZK感知集群状态的变化，并自动重新分配Topic的分区和对应的Broker。
  • Broker控制：Kafka集群中的所有Broker都连接到ZK中。ZK维护了所有活动Broker的列表和状态信息，包括Leader、Follower等信息。如果某个Broker出现故障，ZK可以自动感知它的下线，并通知集群中的其他Broker重新分配Leader。
  • 分布式锁：Zookeeper提供群众同步的机制，使得多个Kafka Broker的协调和管理变得可行。Kafka中的一些操作需要集群中的所有Broker都达成一致意见，因此需要使用ZK协调器的分布式锁机制来维护这些操作的一致性，并防止数据的意外损坏。

Zookeeper快速部署教程可参考我上一篇文章：【中间件】通过 docker-compose 快速部署 Zookeeper 保姆级教程

总之，Zookeeper在Kafka集群中发挥了重要的角色，它管理着Kafka的发布/订阅机制、Broker状态信息、Topic的元数据信息等，使得Kafka集群的分布式协同和协调变得可能。在Kafka写操作（生产者或管理员在Kafka生产或维护上修改了配置）上，ZK用于协作锁定。在Kafka读操作（消费者将订阅的主题分区元数据读取到kafka消费者中）上，ZK用于协作。

五、Kafka 编排部署

1）下载 Kafka

  1. wget https://downloads.apache.org/kafka/3.4.0/kafka_2.12-3.4.0.tgz --no-check-certificate

注意还需要java环境，可以去官网下载，也可以在我下面提供的地址下载：

链接: https://pan.baidu.com/s/1o_z3t16v0eASYWN4VcjYeg?pwd=kuac 提取码: 

  1. kuac

2）配置

  • config/kafka-node1/server.properties
  1. # 常见配置挂载目录mkdir config/{kafka-node1,kafka-node2,kafka-node3} -p
  3. # 配置cat>config/kafka-node1/server.properties<<EOF
  4. #broker的全局唯一编号，不能重复
  5. broker.id=1
  7. #删除topic功能使能
  8. delete.topic.enable=true
  9. #处理网络请求的线程数量
  10. num.network.threads=3
  11. #用来处理磁盘IO的现成数量
  12. num.io.threads=8
  13. #发送套接字的缓冲区大小
  14. socket.send.buffer.bytes=102400
  15. #接收套接字的缓冲区大小
  16. socket.receive.buffer.bytes=102400
  17. #请求套接字的缓冲区大小
  18. socket.request.max.bytes=104857600
  19. #kafka数据的存储位置
  20. log.dirs=/opt/apache/kafka/logs
  21. #指定Topic的分区数量，这里设置为3。 默认只有一个分区，设置多分区可以支持并发读写和负载均衡
  22. num.partitions=3
  23. #副本，默认只有一个副本，不会进行数据备份和冗余
  24. replication.factor=3
  25. #用来恢复和清理data下数据的线程数量
  26. num.recovery.threads.per.data.dir=1
  27. #segment文件保留的最长时间，超时将被删除
  28. log.retention.hours=168
  29. #配置连接Zookeeper集群地址
  30. zookeeper.connect=zookeeper-node1:2181,zookeeper-node2:2181,zookeeper-node3:2181
  31. #zookeeper连接超时时间
  32. zookeeper.connection.timeout.ms=60000
  33. EOF
  • config/kafka-node2/server.properties
  1. cat>config/kafka-node2/server.properties<<EOF
  2. #broker的全局唯一编号，不能重复
  3. broker.id=2
  5. #删除topic功能使能
  6. delete.topic.enable=true
  7. #处理网络请求的线程数量
  8. num.network.threads=3
  9. #用来处理磁盘IO的现成数量
  10. num.io.threads=8
  11. #发送套接字的缓冲区大小
  12. socket.send.buffer.bytes=102400
  13. #接收套接字的缓冲区大小
  14. socket.receive.buffer.bytes=102400
  15. #请求套接字的缓冲区大小
  16. socket.request.max.bytes=104857600
  17. #kafka数据的存储位置
  18. log.dirs=/opt/apache/kafka/logs
  19. #指定Topic的分区数量，这里设置为3。 默认只有一个分区，设置多分区可以支持并发读写和负载均衡
  20. num.partitions=3
  21. #副本，默认只有一个副本，不会进行数据备份和冗余
  22. replication.factor=3
  23. #用来恢复和清理data下数据的线程数量
  24. num.recovery.threads.per.data.dir=1
  25. #segment文件保留的最长时间，超时将被删除
  26. log.retention.hours=168
  27. #配置连接Zookeeper集群地址
  28. zookeeper.connect=zookeeper-node1:2181,zookeeper-node2:2181,zookeeper-node3:2181
  29. #zookeeper连接超时时间
  30. zookeeper.connection.timeout.ms=60000
  31. EOF
  • config/kafka-node3/server.properties
  1. cat>config/kafka-node3/server.properties<<EOF
  2. #broker的全局唯一编号，不能重复
  3. broker.id=3
  5. #删除topic功能使能
  6. delete.topic.enable=true
  7. #处理网络请求的线程数量
  8. num.network.threads=3
  9. #用来处理磁盘IO的现成数量
  10. num.io.threads=8
  11. #发送套接字的缓冲区大小
  12. socket.send.buffer.bytes=102400
  13. #接收套接字的缓冲区大小
  14. socket.receive.buffer.bytes=102400
  15. #请求套接字的缓冲区大小
  16. socket.request.max.bytes=104857600
  17. #kafka数据的存储位置
  18. log.dirs=/opt/apache/kafka/logs
  19. #指定Topic的分区数量，这里设置为3。 默认只有一个分区，设置多分区可以支持并发读写和负载均衡
  20. num.partitions=3
  21. #副本，默认只有一个副本，不会进行数据备份和冗余
  22. replication.factor=3
  23. #用来恢复和清理data下数据的线程数量
  24. num.recovery.threads.per.data.dir=1
  25. #segment文件保留的最长时间，超时将被删除
  26. log.retention.hours=168
  27. #配置连接Zookeeper集群地址
  28. zookeeper.connect=zookeeper-node1:2181,zookeeper-node2:2181,zookeeper-node3:2181
  29. #zookeeper连接超时时间
  30. zookeeper.connection.timeout.ms=60000
  31. EOF

3）启动脚本 bootstrap.sh

  1. #!/usr/bin/env sh${KAFKA_HOME}/bin/kafka-server-start.sh ${KAFKA_HOME}/config/server.properties

4）构建镜像 Dockerfile

  1. FROM registry.cn-hangzhou.aliyuncs.com/bigdata_cloudnative/centos:7.7.1908
  3. RUN rm -f /etc/localtime &&ln -sv /usr/share/zoneinfo/Asia/Shanghai /etc/localtime &&echo"Asia/Shanghai"> /etc/timezone
  5. RUN exportLANG=zh_CN.UTF-8
  7. # 创建用户和用户组，跟yaml编排里的user: 10000:10000
  8. RUN groupadd --system --gid=10000 hadoop &&useradd --system --home-dir /home/hadoop --uid=10000 --gid=hadoop hadoop -m
  10. # 安装sudo
  11. RUN yum -y installsudo;chmod640 /etc/sudoers
  13. # 给hadoop添加sudo权限
  14. RUN echo"hadoop ALL=(ALL) NOPASSWD: ALL">> /etc/sudoers
  16. RUN yum -y installinstall net-tools telnet wgetncless
  18. RUN mkdir /opt/apache/
  20. # 添加配置 JDK
  21. ADD jdk-8u212-linux-x64.tar.gz /opt/apache/
  22. ENV JAVA_HOME /opt/apache/jdk1.8.0_212
  23. ENV PATH$JAVA_HOME/bin:$PATH# 添加配置 kafka server
  24. ENV KAFKA_VERSION 2.12-3.4.0
  25. ADD kafka_${KAFKA_VERSION}.tgz /opt/apache/
  26. ENV KAFKA_HOME /opt/apache/kafka
  27. RUN ln -s /opt/apache/kafka_${KAFKA_VERSION}-bin $KAFKA_HOME# 创建数据存储目录
  28. RUN mkdir -p ${KAFKA_HOME}/data/logs
  30. # copy bootstrap.sh
  31. COPY bootstrap.sh /opt/apache/
  32. RUN chmod +x /opt/apache/bootstrap.sh
  34. RUN chown -R hadoop:hadoop /opt/apache
  36. WORKDIR $KAFKA_HOME

开始构建镜像

  1. # 需要查看构建镜像详细过程则需要加上 --progress=plain 选项
  2. docker build -t registry.cn-hangzhou.aliyuncs.com/bigdata_cloudnative/kafka:2.12-3.4.0 . --no-cache --progress=plain
  4. # 为了方便小伙伴下载即可使用，我这里将镜像文件推送到阿里云的镜像仓库
  5. docker push registry.cn-hangzhou.aliyuncs.com/bigdata_cloudnative/kafka:2.12-3.4.0
  7. ### 参数解释# -t：指定镜像名称# . ：当前目录Dockerfile# -f：指定Dockerfile路径#  --no-cache：不缓存

5）编排 docker-compose.yaml

  1. version: '3'
  2. services:
  3.   kafka-node1:
  4.     image: registry.cn-hangzhou.aliyuncs.com/bigdata_cloudnative/kafka:2.12-3.4.0
  5.     user: "hadoop:hadoop"
  6.     container_name: kafka-node1
  7.     hostname: kafka-node1
  8.     restart: always
  9.     privileged: true
  10.     env_file:
  11.       - .env
  12.     volumes:
  13.       - ./config/kafka-node1/server.properties:${KAFKA_HOME}/config/server.properties
  14.     ports:
  15.       - "${KAFKA_NODE1_SERVER_PORT}:9092"
  16.     expose:
  17.       - 2888
  18.       - 3888
  19.     command: ["sh","-c","/opt/apache/bootstrap.sh"]
  20.     networks:
  21.       - hadoop-network
  22.     healthcheck:
  23.       test: ["CMD-SHELL", "netstat -tnlp|grep :9092 || exit 1"]
  24.       interval: 10s
  25.       timeout: 10s
  26.       retries: 5
  27.   kafka-node2:
  28.     image: registry.cn-hangzhou.aliyuncs.com/bigdata_cloudnative/kafka:2.12-3.4.0
  29.     user: "hadoop:hadoop"
  30.     container_name: kafka-node2
  31.     hostname: kafka-node2
  32.     restart: always
  33.     privileged: true
  34.     env_file:
  35.       - .env
  36.     volumes:
  37.       - ./config/kafka-node2/server.properties:${KAFKA_HOME}/config/server.properties
  38.     ports:
  39.       - "${KAFKA_NODE2_SERVER_PORT}:9092"
  40.     expose:
  41.       - 2888
  42.       - 3888
  43.     command: ["sh","-c","/opt/apache/bootstrap.sh"]
  44.     networks:
  45.       - hadoop-network
  46.     healthcheck:
  47.       test: ["CMD-SHELL", "netstat -tnlp|grep :9092 || exit 1"]
  48.       interval: 10s
  49.       timeout: 10s
  50.       retries: 5
  51.   kafka-node3:
  52.     image: registry.cn-hangzhou.aliyuncs.com/bigdata_cloudnative/kafka:2.12-3.4.0
  53.     user: "hadoop:hadoop"
  54.     container_name: kafka-node3
  55.     hostname: kafka-node3
  56.     restart: always
  57.     privileged: true
  58.     env_file:
  59.       - .env
  60.     volumes:
  61.       - ./config/kafka-node3/server.properties:${KAFKA_HOME}/config/server.properties
  62.     ports:
  63.       - "${KAFKA_NODE3_SERVER_PORT}:9092"
  64.     expose:
  65.       - 2888
  66.       - 3888
  67.     command: ["sh","-c","/opt/apache/bootstrap.sh"]
  68.     networks:
  69.       - hadoop-network
  70.     healthcheck:
  71.       test: ["CMD-SHELL", "netstat -tnlp|grep :9092 || exit 1"]
  72.       interval: 10s
  73.       timeout: 10s
  74.       retries: 5# 连接外部网络
  75. networks:
  76.   hadoop-network:
  77.     external: true
  1. .env

环境变量文件内容如下：

  1. # 对外暴露的端口cat<<EOF> .env
  2. KAFKA_HOME=/opt/apache/kafka
  3. KAFKA_NODE1_SERVER_PORT=39092
  4. KAFKA_NODE2_SERVER_PORT=39093
  5. KAFKA_NODE3_SERVER_PORT=39094
  6. EOF

6）开始部署

  1. docker-compose -f docker-compose.yaml up -d
  3. # 查看
  4. docker-compose -f docker-compose.yaml ps

六、简单测试验证

  1. # 登录zookeeper，在zookeeper查看brokers${ZOOKEEPER_HOME}/bin/zkCli.sh ls /brokers/ids
  3. ${ZOOKEEPER_HOME}/bin/zkCli.sh get /brokers/ids/1
  4. ${ZOOKEEPER_HOME}/bin/zkCli.sh get /brokers/ids/2
  5. ${ZOOKEEPER_HOME}/bin/zkCli.sh get /brokers/ids/3

七、常用的 Kafka 客户端命令

1）添加topic

  1. # 随便登录
  2. docker exec -it kafka-node1 bash# 创建topic，1副本，1分区，设置数据过期时间72小时（-1表示不过期），单位ms，72*3600*1000=259200000${KAFKA_HOME}/bin/kafka-topics.sh --create --topic test002 --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092  --partitions 1 --replication-factor 1 --config retention.ms=259200000

2）查看topic

  1. # 查看topic列表${KAFKA_HOME}/bin/kafka-topics.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --list
  3. # 查看topic列表详情${KAFKA_HOME}/bin/kafka-topics.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --describe
  5. # 指定topic${KAFKA_HOME}/bin/kafka-topics.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --describe --topic test002
  7. # 查看消费者组${KAFKA_HOME}/bin/kafka-consumer-groups.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --list
  8. kafka-consumer-groups.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --describe  --group test002

3）修改topic

  1. # 修改分区，扩分区，不能减少分区${KAFKA_HOME}/bin/kafka-topics.sh --alter --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --topic test002 --partitions 2# 修改过期时间，下面两行都可以${KAFKA_HOME}/bin/kafka-configs.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --alter --topic test002 --add-config retention.ms=86400000${KAFKA_HOME}/bin/kafka-configs.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --alter --entity-name test002 --entity-type topics --add-config retention.ms=86400000# 修改副本数，将副本数修改成3
  2. $ cat>1.json<<EOF
  3. {"version":1,
  4. "partitions":[
  5. {"topic":"test002","partition":0,"replicas":[0,1,2]},
  6. {"topic":"test002","partition":1,"replicas":[1,2,0]},
  7. {"topic":"test002","partition":2,"replicas":[2,0,1]}
  8. ]}
  9. EOF${KAFKA_HOME}/bin/kafka-topics.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --describe --topic test002

4）扩容分区

  1. #把test002 topic扩容为6个分区。#注意：目前不支持减少分区，扩容前必须存在这个主题。${KAFKA_HOME}/bin/kafka-topics.sh -alter --partitions 6 --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --topic test002
  3. ${KAFKA_HOME}/bin/kafka-topics.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --describe

在这里插入图片描述

5）删除topic

  1. ${KAFKA_HOME}/bin/kafka-topics.sh --delete --topic test002 --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092

6）生成者和消费者

生产者

  1. ${KAFKA_HOME}/bin/kafka-console-producer.sh --broker-list kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --topic test002
  2. {"id":"1","name":"n1","age":"20"}{"id":"2","name":"n2","age":"21"}{"id":"3","name":"n3","age":"22"}

消费者

  1. # 从头开始消费${KAFKA_HOME}/bin/kafka-console-consumer.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --topic test002 --from-beginning
  3. # 指定从分区的某个位置开始消费，这里只指定了一个分区，可以多写几行或者遍历对应的所有分区${KAFKA_HOME}/bin/kafka-console-consumer.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --topic test002 --partition 0 --offset 100

7）消费组

在 Kafka 中，消费组（Consumer Group）是一组独立消费者的集合，它们共同消费一个或多个 Topic 中的数据。消费组内的消费者协同工作，通过分摊该 Topic 中的所有分区，以实现消息的消费和处理。

消费组在 Kafka 消息队列中起到了至关重要的作用。它可以提供如下功能：

  • 并发消费：消费组内的每个消费者都可以独立地消费消息，可以实现高并发处理。
  • 自动负载均衡：消费组内的消费者会自动协作，将消费任务均分到所有消费者上，使得每个消费者都能处理相同数量的消息。
  • 提高可用性：当消费组内的一个或多个消费者故障退出时，消息会自动分配到其他消费者上，保证消费任务的不间断执行。
  • 支持多租户：可以通过 Consumer Group 来对不同的租户进行消息隔离，不同的 Consumer Group 可以读取同一个 Topic 的不同副本，或者读取不同 Topic 的不同分区，实现多个实例共享同一 Topic 或分散处理不同 Topic。

示例如下：

  1. ${KAFKA_HOME}/bin/kafka-console-consumer.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --topic test002 --group test002

通常情况下，消费组中的消费者都运行在不同的机器上，这样就可以实现分布式消费，以提高消息处理性能和可用性。Kafka 对消费组的实现也非常简单，通过在消费者在订阅 Topic 时，接受一个 Group ID 参数，就可以自动加入到一个消费组中。Kafka 会将Group ID 相同的消费者映射到同一个 Consumer Group 中，以实现协同消费和分摊消费任务的目的。

8）查看数据积压

  1. ${KAFKA_HOME}/bin/kafka-consumer-groups.sh --bootstrap-server kafka-node1:9092,kafka-node2:9092,kafka-node3:9092 --describe --group test002

9）kafka 数据积压处理方法

在 Kafka 中，由于消息的生产和消费速度可能不一致，导致消息会积压在 Kafka 的分区中，如果这些积压的消息处理不及时，会导致 Kafka 系统的性能下降和可用性降低等问题。因此，需要采取一些处理方法来解决数据积压问题：

  • 增加消费者：增加消费者可以使消费任务并行执行，加快消息的处理速度。可以通过增加消费者的方式将积压的消息消费掉，提高系统处理速度和效率。
  • 调整消费者组：当一个消费组中的消费者无法处理所有的消息时，可以考虑调整消费者组。可以增加消费者的数量或者更换消费者组，以适应消息处理的速度和大小。
  • 调整消息分区：Kafka 中Topic 的分区数也会影响数据积压的情况。可以调整分区数以改善数据读取和分发的情况，或者对热点 Topic 进行分区处理，以实现更好的性能和可用性。
  • 调整消费 offset：若积压的消息都已经被处理过了，却还在 Kafka 中存在，可能是消费者消费 offset 设置错误导致的。可以通过 Kafka 的 offset 操作，重置消费 offset，跳过已经处理过的消息，减少数据积压的问题。
  • 执行消息清洗：在消费 Kafka 消息时，可以额外执行一些消息清洗处理操作，将无用的数据过滤出去，或者将数据进行清理和格式化处理，减少中间处理环节，提高数据消费的效率和可用性。

以上是一些解决 Kafka 数据积压问题的常用方法，需要视具体情况而定，选择合适的方法来解决。

通过 docker-compose 快速部署 Kafka 教程就先到这里了，有任何疑问欢迎给我留言或私信，可关注我公众号【大数据与云原生技术分享】加群交流或私信沟通~

标签： kafka 中间件 docker
本文转载自: https://blog.csdn.net/qq_35745940/article/details/130788865
版权归原作者 大数据老司机 所有， 如有侵权，请联系我们删除。
登录后发布评论

“【中间件】通过 docker-compose 快速部署 Kafka 保姆级教程”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？