基于云开发者空间与容器技术的分布式消息中间件Kafka部署与应用

基于云开发者空间与容器技术的分布式消息中间件Kafka部署与应用

学习目标

  1. 理解分布式消息中间件Kafka的基础架构、工作原理；
  2. 掌握在不同硬件平台和操作系统架构上搭建与之适配的容器环境；
  3. 掌握容器化技术一键部署Kafka依赖组件、Kafka集群以及可视化管理应用；
  4. 掌握华为云主机开发者空间的CodeArts IDE进行代码、文件编辑；
  5. 能够基于Python编写自定义Kafka生产者、消费者应用程序；
  6. 能够基于容器化技术对Python程序进行容器镜像打包；
  7. 掌握容器化技术一键部署基于Python的Kafka生产者、消费者应用程序；
  8. 能够基于可视化管理应用理解Kafka消息的处理流程。

目录

  1. 概述
    • 1.1 课程实验内容
  2. 系统原理
    • 2.1 Kafka基础架构及相关名词解释
    • 2.2 Kafka工作原理
  3. 系统架构设计与实现
    • 3.1 系统架构
    • 3.2 系统技术选型
    • 3.3 系统功能模块设计与实现
  4. 工具安装与应用部署
    • 4.1 Docker与Docker Compose安装配置
    • 4.2 Kafka集群容器化部署
    • 4.3 Python生产者与消费者程序开发
    • 4.4 容器镜像构建与运行
    • 4.5 使用docker-compose一键部署生产者与消费者
  5. 测试评估
    • 5.1 Kafka集群与管理平台测试
    • 5.2 生产者与消费者功能测试
  6. 遇到的问题与解决过程
  7. 参考文献

第一部分：系统原理

1. 分布式中间件 Kafka

Kafka基础架构

图 1 Kafka基础架构

Kafka核心组件说明

  • Producer（生产者）：消息的产生者，是消息的入口。

  • Broker：Kafka实例，每个服务器上有一个或多个Kafka的实例。每个Kafka集群内的Broker都有一个不重复的编号，如图中的broker-0、broker-1等。

  • Topic（主题）：消息的主题，可以理解为消息的分类，Kafka的数据就保存在Topic。在每个Broker上都可以创建多个Topic。

  • Partition（分区）：Topic的分区，每个Topic可以有多个分区，分区的作用是做负载，提高Kafka的吞吐量。同一个Topic在不同的分区的数据是不重复的，Partition的表现形式就是一个一个的文件夹！

  • Replication（副本）：每一个分区都有多个副本，副本的作用是做备胎。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在Kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，Follower和Leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。

  • Message（消息）：每一条发送的消息主体。

  • Consumer（消费者）：消息的消费方，是消息的出口。

  • Consumer Group（消费者组）：可以将多个消费组组成一个消费者组，在Kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个Topic的不同分区的数据，这也是为了提高Kafka的吞吐量！

  • Zookeeper：Kafka集群依赖Zookeeper来保存集群的元信息，来保证系统的可用性。

2. Kafka工作原理

① 发送数据

Kafka工作流程

图 2 Kafka工作流程

消息写入Leader后，Follower是主动的去Leader进行同步的。Producer采用push模式将数据发布到Broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的。

顺序写入分区

图 3 顺序写入分区

② 保存数据

Producer将数据写入Kafka后，集群就需要对数据进行保存了。Kafka初始会单独开辟一块磁盘空间，顺序写入数据（效率比随机写入高）。

每个Topic都可以分为一个或多个Partition，Partition在服务器上的表现形式就是一个一个的文件夹，每个Partition的文件夹下面会有多组segment文件，每组segment文件又包含.index文件、.log文件、.timeindex文件三个文件，log文件就实际是存储message的地方，而index和timeindex文件为索引文件，用于检索消息。

分区结构

图 4 分区结构

③ 消费数据

消息存储在log文件后，消费者就可以进行消费了。Kafka采用的是发布订阅模式，消费者主动的去Kafka集群拉取消息，与Producer相同的是，消费者在拉取消息的时候也是找Leader去拉取。

多个消费者可以组成一个消费者组（consumer group），每个消费者组都有一个组id！同一个消费组者的消费者可以消费同一Topic下不同分区的数据，但是不会组内多个消费者消费同一分区的数据。

消费消息分区中的消息

图 5 消费消息分区中的消息

3. 容器化技术

① Docker技术架构

Docker技术架构

图 6 Docker技术架构

② 基于Docker的镜像管理

基于Docker的镜像管理

图 7 基于Docker的镜像管理

③ Docker容器生命周期管理与状态转换

Docker容器生命周期管理与状态转换

图 8 Docker容器生命周期管理与状态转换

与容器镜像、容器相关的一些操作或命令，可参考如下网址进行学习：

https://www.runoob.com/docker/docker-tutorial.html

第二部分：工具安装与应用部署

第一步：在华为云开发者空间创建环境（5分钟）

  1. 登录 华为云开发者空间
  2. 点击「新建开发环境」→ 选择：
    • 规格：4vCPU 8GB（手册要求）
    • 操作系统：Ubuntu 24.04
    • 镜像：推荐用「基础 Ubuntu」或「Docker 已预装」模板
  3. 等待环境启动，进入 CodeArts IDE（网页版终端）

第二步：安装 Docker + docker-compose（环境准备）

在终端里依次复制下面所有命令（一行一行执行）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# 1. 更新apt包索引
sudo apt-get update

# 2. 安装apt依赖包，用于通过HTTPS来获取仓库
sudo apt-get install \
    apt-transport-https \
    ca-certificates \
    curl \
    gnupg-agent \
    software-properties-common

# 3. 添加Docker的官方GPG密钥或华为云镜像的Docker GPG密钥文件
sudo curl -fsSL https://mirrors.ustc.edu.cn/docker-ce/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc
sudo curl -fsSL https://repo.huaweicloud.com/docker-ce/linux/ubuntu/gpg -o /etc/apt/keyrings/docker.asc

# 4. 设置仓库
sudo tee /etc/apt/sources.list.d/docker.list > /dev/null <<EOF
deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.asc] https://repo.huaweicloud.com/docker-ce/linux/ubuntu/ noble stable
EOF

# 5. 再次更新apt包索引
sudo apt-get update

# 6. 安装 Docker（指定版本）
VERSION_STRING=5:26.0.1-1~ubuntu.24.04~noble
sudo apt-get install docker-ce=$VERSION_STRING docker-ce-cli=$VERSION_STRING containerd.io

# 7. 启动 Docker
sudo systemctl start docker
sudo systemctl enable docker

# 8. 验证
sudo docker version

安装docker-compose

若出现拉取缓慢则安装docker compose：

1
2
# 下载arm架构（华为鲲鹏）的docker-compose
sudo curl -SL https://github.com/docker/compose/releases/download/v2.40.3/docker-compose-linux-aarch64 -o /usr/bin/docker-compose

其它版本：

若出现拉取缓慢：

1
2
sudo apt-get install -y docker-compose-plugin
docker compose version

第三步：拉取 Kafka 镜像 + 创建配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# 配置 Docker 加速器（防止后面拉镜像慢）
sudo tee /etc/docker/daemon.json > /dev/null <<EOF
{
  "registry-mirrors": [
    "https://8080000058877000000.mirror.swr.myhuaweicloud.com",
    "https://docker.registry.cyou",
    "https://mirror.aliyuncs.com"
  ]
}
EOF

sudo systemctl restart docker

# 登录华为 SWR（提供的账号密码）
sudo docker login -u cn-east-3@HPUANMUGSPDRGE7QSJBR -p 3bc607b6f7d7510008b9a734b542cd15968195396828b018cd2889dc050fd698 swr.cn-east-3.myhuaweicloud.com

# 拉取并打标签（严格按实验手册）
sudo docker pull swr.cn-east-3.myhuaweicloud.com/whucs/zookeeper
sudo docker pull swr.cn-east-3.myhuaweicloud.com/whucs/wurstmeister/kafka
sudo docker pull swr.cn-east-3.myhuaweicloud.com/whucs/provectuslabs/kafka-ui

sudo docker tag swr.cn-east-3.myhuaweicloud.com/whucs/zookeeper zookeeper
sudo docker tag swr.cn-east-3.myhuaweicloud.com/whucs/wurstmeister/kafka wurstmeister/kafka
sudo docker tag swr.cn-east-3.myhuaweicloud.com/whucs/provectuslabs/kafka-ui provectuslabs/kafka-ui

# 检查镜像是否拉取成功
sudo docker images | grep -E 'zookeeper|kafka|kafka-ui'

1. 创建日志文件夹

1
mkdir -p kafka1-logs kafka2-logs kafka3-logs

2. 创建 docker-compose-kafka-new.yml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
cat > docker-compose-kafka-new.yml << 'EOF'
version: '3.5'
services:
  zookeeper:
    image: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
  kafka1:
    image: wurstmeister/kafka
    container_name: kafka1
    ports:
      - 9091:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.67.17.234:9091
      KAFKA_CREATE_TOPICS: "kafeidou:2:0"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
    volumes:
      - ./kafka1-logs:/kafka
    depends_on:
      - zookeeper
  kafka2:
    image: wurstmeister/kafka
    container_name: kafka2
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.67.17.234:9092
      KAFKA_CREATE_TOPICS: "kafeidou:2:0"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
    volumes:
      - ./kafka2-logs:/kafka
    depends_on:
      - zookeeper
  kafka3:
    image: wurstmeister/kafka
    container_name: kafka3
    ports:
      - 9093:9092
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.67.17.234:9093
      KAFKA_CREATE_TOPICS: "kafeidou:2:0"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
    volumes:
      - ./kafka3-logs:/kafka
    depends_on:
      - zookeeper
  kafka-manager:
    image: provectuslabs/kafka-ui
    container_name: kafka-manager
    environment:
      DYNAMIC_CONFIG_ENABLED: 'true'
    ports:
      - "9009:8080"
EOF

3. 一键启动 Kafka 集群 + 可视化界面

1
2
sudo docker compose -f docker-compose-kafka-new.yml up -d
sudo docker ps

Kafka集群启动

Docker容器状态

4. 访问Kafka管理平台

华为云服务器中firefox浏览器访问：http://localhost:9009 或 http://10.67.17.234:9009 即可看到Kafka平台UI搭建完成：

Kafka UI界面

添加/修改集群，自定义命名kafka-cluster，关联zookeeper服务器，点击submit确定进入：

配置集群

5. 在Kafka集群中创建Topic

（1）在任意一个 Kafka 容器里执行：

1
sudo docker exec -it kafka1 /bin/bash

（2）使用命令创建Topic，并检查Topic是否创建成功：

1
2
3
4
5
kafka-topics.sh --create \
--topic my_topic \
--partitions 3 \
--replication-factor 1 \
--bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092

第四步：Python 生产者 + 消费者 + 打包成镜像

1. 创建项目目录和producer.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 创建项目目录
mkdir -p kafka-app && cd kafka-app

# 创建 producer.py
cat > producer.py << 'EOF'
from kafka import KafkaProducer
import json
import os

def send_message(producer, topic, message):
    producer.send(topic, message)

if __name__ == "__main__":
    kafka_servers = os.getenv('KAFKA_SERVERS', '10.67.17.234:9091,10.67.17.234:9092,10.67.17.234:9093')
    topic = os.getenv('TOPIC','test-topic')
    num_messages = int(os.getenv('COUNT',10000))

    producer = KafkaProducer(
        bootstrap_servers=kafka_servers.split(','),
        value_serializer=lambda m: json.dumps(m).encode('utf-8')
    )

    for i in range(num_messages):
        message = {'id': i, 'message': f'This is message {i}'}
        send_message(producer, topic, message)
        print(f"Sent message {i}")
    producer.flush()
EOF

2. 本地测试运行

1
2
3
4
5
6
7
8
9
# 创建虚拟环境（建议放在项目文件夹里）
python3 -m venv ~/myenv
# 激活虚拟环境
source ~/myenv/bin/activate
# 安装依赖
pip install kafka-python

# 运行测试
python producer.py

3. 创建生产者Dockerfile

1
2
3
4
5
6
7
8
9
10
11
12
cat > Dockerfile << 'EOF'
# 使用官方 Python 镜像作为基础镜像
FROM python:3.8
# 设置工作目录
WORKDIR /app
# 将当前目录内容复制到镜像中
COPY ./producer.py /app/
# 安装依赖（使用清华镜像加速）
RUN pip install kafka-python -i https://pypi.tuna.tsinghua.edu.cn/simple
# 运行生产者脚本
CMD ["python", "./producer.py"]
EOF

4. 构建生产者 Docker 镜像

1
sudo docker build -t python-kafka-producer .

5. 运行生产者容器

1
sudo docker run --name kafka-producer python-kafka-producer

6. 创建消费者 consumer.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
cat > consumer.py << 'EOF'
from kafka import KafkaConsumer
import json
import os

def consume_messages(consumer, topic):
    for message in consumer:
        print(f"Received message: {message.value.decode('utf-8')}")

if __name__ == "__main__":
    kafka_servers = os.getenv('KAFKA_SERVERS', '10.67.17.234:9091,10.67.17.234:9092,10.67.17.234:9093')
    topic = os.getenv('TOPIC','test-topic')

    consumer = KafkaConsumer(
        topic,
        bootstrap_servers=kafka_servers.split(','),
        auto_offset_reset='earliest',
        enable_auto_commit=True,
        group_id='my-group'
    )
    consume_messages(consumer, topic)
EOF

7. 创建消费者专用Dockerfile

1
2
3
4
5
6
7
cat > Dockerfile.consumer << 'EOF'
FROM python:3.8
WORKDIR /app
COPY ./consumer.py /app/
RUN pip install kafka-python -i https://pypi.tuna.tsinghua.edu.cn/simple
CMD ["python", "./consumer.py"]
EOF

8. 构建消费者镜像并运行

1
2
3
4
5
# 构建消费者镜像
sudo docker build -f Dockerfile.consumer -t python-kafka-consumer .

# 运行消费者容器
sudo docker run --name kafka-consumer python-kafka-consumer

第五步：用 docker-compose 一键运行生产者 + 消费者

1. 清理旧容器（防止端口冲突）

1
2
sudo docker stop kafka-producer kafka-consumer 2>/dev/null || true
sudo docker rm kafka-producer kafka-consumer 2>/dev/null || true

2. 创建 docker-compose.yml（一次性启动生产者和消费者）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
cat > docker-compose.yml << 'EOF'
version: '3'
services:
  kafka-producer:
    build:
      context: .
      dockerfile: Dockerfile
    image: python-kafka-producer:latest
    environment:
      - KAFKA_SERVERS=10.67.17.234:9091,10.67.17.234:9092,10.67.17.234:9093
      - TOPIC=test-topic-new
      - COUNT=5000
    restart: on-failure

  kafka-consumer:
    build:
      context: .
      dockerfile: Dockerfile.consumer
    image: python-kafka-consumer:latest
    environment:
      - KAFKA_SERVERS=10.67.17.234:9091,10.67.17.234:9092,10.67.17.234:9093
      - TOPIC=test-topic-new
    depends_on:
      - kafka-producer
    restart: on-failure
EOF

3. 一键启动

1
sudo docker compose -f docker-compose.yml up -d

4. 查看实时日志

1
sudo docker-compose -f docker-compose.yml logs -f

看到生产者和消费者都在工作就成功了！

第三部分：测试评估

5.1 Kafka集群与管理平台测试

  • 验证Kafka集群是否正常运行
  • 检查Zookeeper连接状态
  • 测试Kafka UI管理平台功能

5.2 生产者与消费者功能测试

  • 验证生产者能够成功发送消息
  • 验证消费者能够成功接收消息
  • 检查消息的完整性和顺序性

第四部分：遇到的问题与解决过程

问题1：Docker镜像拉取缓慢

解决方案：配置Docker加速器，使用华为云SWR镜像源。

问题2：ARM架构兼容性

解决方案：使用华为鲲鹏专用的docker-compose版本。

问题3：Kafka连接问题

解决方案：确保KAFKA_ADVERTISED_LISTENERS配置正确，使用正确的IP地址和端口。

参考文献

  1. Kafka官方文档：https://kafka.apache.org/documentation/
  2. Docker官方文档：https://docs.docker.com/
  3. 华为云开发者空间：https://developer.huaweicloud.com/devspace
  4. Docker教程：https://www.runoob.com/docker/docker-tutorial.html