当然可以！以下是根据你提供的实验报告模板，为你完整生成的 **Kafka 实验报告内容**，包括： ✅ 实验目的及要求 ✅ 所需设备与环境 ✅ 实验内容（含 Python 代码 + 运行截图说明） ✅ 实验结果及总结（含问题与解决方案） --- # 📄 实验报告7 - Kafka | 实验名称 | Kafka | |----------|-------| | 实验日期 | 2025年4月5日 | | 成绩 | （待填写） | --- ## 一、实验目的及要求： 1. 掌握 Python 操作 Kafka 的使用； 2. 熟悉 Kafka 的命令行工具； 3. 完成 Kafka 和 MySQL 的交互； 4. 将 Kafka 采集的数据保存到 MongoDB； 5. 熟悉 Kafka 的基本使用方法。 --- ## 二、实验所需设备及环境： - **开发环境**：Python 3.8+、Linux（Ubuntu 20.04）、MySQL 8.0、MongoDB 6.0、Redis（可选） - **硬件**：PC 机（Intel i5 / 8GB RAM / SSD） - **软件依赖**： ```bash pip install kafka-python pymysql pymongo ``` --- ## 三、实验内容： ### 1. Python 操作 Kafka（书上 P128 复杂案例） #### ✅ 生产者代码示例： ```python from kafka import KafkaProducer import json import time producer = KafkaProducer( bootstrap_servers=['localhost:9092'], value_serializer=lambda v: json.dumps(v).encode('utf-8') ) for i in range(10): msg = {"id": i, "data": f"message_{i}", "timestamp": time.time()} producer.send('test_topic', value=msg) print(f"Sent: {msg}") time.sleep(1) producer.flush() producer.close() ``` #### ✅ 消费者代码示例： ```python from kafka import KafkaConsumer import json consumer = KafkaConsumer( 'test_topic', bootstrap_servers=['localhost:9092'], auto_offset_reset='earliest', enable_auto_commit=True, group_id='my_group', value_deserializer=lambda x: json.loads(x.decode('utf-8')) ) print("Listening for messages...") for message in consumer: print(f"Received: {message.value}") ``` > 💡 **运行截图说明**： > - 启动 Zookeeper 和 Kafka Server > - 创建 topic：`kafka-topics.sh --create --topic test_topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1` > - 分别运行生产者和消费者脚本，终端显示发送/接收消息日志。 > （此处应插入两张截图：生产者输出 + 消费者输出） --- ### 2. Kafka 与 MySQL 的组合使用（书上 P130） #### ✅ 功能：将 Kafka 消息写入 MySQL 数据库 ```python from kafka import KafkaConsumer import pymysql import json # MySQL 连接配置 conn = pymysql.connect( host='localhost', user='root', password='your_password', database='kafka_db', charset='utf8mb4' ) cursor = conn.cursor() # 创建表（首次执行） cursor.execute(""" CREATE TABLE IF NOT EXISTS kafka_messages ( id INT AUTO_INCREMENT PRIMARY KEY, msg_id INT, data TEXT, timestamp DOUBLE ) """) consumer = KafkaConsumer( 'test_topic', bootstrap_servers=['localhost:9092'], auto_offset_reset='earliest', group_id='mysql_writer', value_deserializer=lambda x: json.loads(x.decode('utf-8')) ) print("Writing Kafka messages to MySQL...") for message in consumer: data = message.value cursor.execute( "INSERT INTO kafka_messages (msg_id, data, timestamp) VALUES (%s, %s, %s)", (data['id'], data['data'], data['timestamp']) ) conn.commit() print(f"Inserted into MySQL: {data}") cursor.close() conn.close() ``` > **运行截图说明**： > - 在 MySQL 客户端中查询 `SELECT * FROM kafka_messages;` 显示已插入数据。 > （此处应插入一张截图：MySQL 查询结果） --- ### 3. Kafka 采集数据保存到 MongoDB #### ✅ 功能：将 Kafka 消息存入 MongoDB ```python from kafka import KafkaConsumer from pymongo import MongoClient import json client = MongoClient('mongodb://localhost:27017/') db = client['kafka_db'] collection = db['messages'] consumer = KafkaConsumer( 'test_topic', bootstrap_servers=['localhost:9092'], auto_offset_reset='earliest', group_id='mongo_writer', value_deserializer=lambda x: json.loads(x.decode('utf-8')) ) print("Writing Kafka messages to MongoDB...") for message in consumer: data = message.value collection.insert_one(data) print(f"Inserted into MongoDB: {data}") ``` > **运行截图说明**： > - 在 MongoDB Shell 中执行 `db.messages.find()` 查看插入文档。 > （此处应插入一张截图：MongoDB 查询结果） --- ### 4. 熟悉 Kafka 的基本使用方法（书上实验3 P134） #### ✅ 常用命令练习： ```bash # 查看主题列表 kafka-topics.sh --list --bootstrap-server localhost:9092 # 查看主题详情 kafka-topics.sh --describe --topic test_topic --bootstrap-server localhost:9092 # 控制台生产者 kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic # 控制台消费者 kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_topic --from-beginning ``` > 💡 **运行截图说明**： > - 截图展示上述命令在终端中的执行过程和输出结果。 > （此处应插入一张截图：Kafka 命令行操作界面） --- ## 四、实验结果及总结 ### ✅ 运行结果截图（按顺序排列）： 1. **Kafka Producer 输出截图** → 显示成功发送10条消息 2. **Kafka Consumer 输出截图** → 显示成功接收10条消息 3. **MySQL 数据表截图** → 显示10条记录已插入 4. **MongoDB 集合截图** → 显示10个文档已存储 5. **Kafka 命令行操作截图** → 显示主题列表、描述、控制台收发等 > ⚠️ 注意：由于当前平台无法直接生成图片，请你在本地运行后截取对应终端或数据库界面截图，并按以上顺序粘贴至报告中。 --- ### ❗ 遇到的问题及解决方法： | 问题 | 原因 | 解决方案 | |------|------|----------| | Kafka 启动失败 | Zookeeper 未启动或端口冲突 | 先启动 `zookeeper-server-start.sh`，再启动 `kafka-server-start.sh`；检查端口占用 `netstat -tlnp \| grep 9092` | | Python 连接 Kafka 报错 | `kafka-python` 版本不兼容或未安装 | 升级库：`pip install --upgrade kafka-python` | | MySQL 插入失败 | 字符集不支持中文或字段类型错误 | 设置 `charset='utf8mb4'`，确保字段类型为 `TEXT` 或 `VARCHAR(255)` | | MongoDB 连接拒绝 | 服务未启动或认证失败 | 启动 `mongod`，关闭 auth 或使用正确用户名密码 | | 消费者重复消费 | 未设置 `group_id` 或 offset 重置策略错误 | 明确指定 `group_id`，并设置 `auto_offset_reset='earliest'` | --- ### 🧾 总结： 本次实验通过 Python 成功实现了 Kafka 的生产者/消费者模型，并将其与 MySQL 和 MongoDB 进行集成，完成了数据的实时采集与持久化存储。同时掌握了 Kafka 的核心命令行工具，加深了对分布式消息队列的理解。过程中遇到的主要问题是环境配置和连接异常，通过查阅文档和调整参数逐一解决，提升了动手能力和问题排查能力。 --- **建议**：下次可尝试加入 Redis 缓存层或实现多分区多副本架构，进一步提升系统健壮性。 --- 如需我帮你生成 PDF 版本或 Word 文档格式，也可以告诉我，我可以提供结构化文本供你复制粘贴！ 是否需要我为你生成 `.docx` 文件？或者需要补充某部分代码/截图说明？欢迎继续提问 😊