Kafka 生产消费测试
本文讲述如何利用 Python 先定时向 Kafka 写入数据，再定时从 Kafka 取出数据进行下一步处理（示例写入指定文件），仅供参考学习。
测试过程
- 第1步，登录服务器（示例 192.168.31.101），先安装 Kafka 3.5.0（过程省略）再创建一个 topic，示例 mytopic
#.删除topic
/data/kafka_9092/bin/kafka-topics.sh --bootstrap-server 192.168.31.101:9092 --delete --topic mytopic
#.创建topic
/data/kafka_9092/bin/kafka-topics.sh --bootstrap-server 192.168.31.101:9092 --create --replication-factor 1 --partitions 3 --topic mytopic
#.查看topic
/data/kafka_9092/bin/kafka-topics.sh --bootstrap-server 192.168.31.101:9092 --describe --topic mytopic
- 第2步，安装 Python 3.12.3
curl -sL 'http://iso.sqlfans.cn/python/install_python_3123.sh' | bash
python3 --version
- 第3步，安装 kafka-python
注：未安装 kafka-python 执行 python3 kafka_write.py 会报错 ModuleNotFoundError: No module named 'kafka'
pip3 install kafka-python
- 第4步，利用 Python 每5秒向 Kafka 写入一条数据，将下面脚本保存为 kafka_write.py
import time
import json
import random
import warnings
from datetime import datetime
from kafka import KafkaProducer
from kafka.errors import KafkaError
warnings.filterwarnings("ignore")
# Kafka配置
KAFKA_BOOTSTRAP_SERVERS = '192.168.31.101:9092'
TOPIC_NAME = 'mytopic'
def create_producer():
try:
producer = KafkaProducer(
bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
value_serializer=lambda v: json.dumps(v).encode('utf-8'),
key_serializer=lambda v: v.encode('utf-8') if v else None,
acks='all',
retries=3,
max_block_ms=5000
)
return producer
except Exception as e:
print(f"创建生产者失败: {e}")
return None
def generate_data():
return {
'timestamp': datetime.now().isoformat(),
'message_id': random.randint(1000, 9999),
'temperature': round(random.uniform(20.0, 35.0), 2),
'humidity': round(random.uniform(40.0, 80.0), 2),
'status': random.choice(['normal', 'warning', 'error']),
'data': f"sample_data_{random.randint(1, 100)}"
}
def send_data_kafka(producer, topic, data):
try:
future = producer.send(topic, value=data)
record_metadata = future.get(timeout=10)
print(f"数据已发送 - 分区: {record_metadata.partition}, "
f"偏移量: {record_metadata.offset}, "
f"数据: {data}")
return True
except KafkaError as e:
print(f"Error发送失败: {e}")
return False
except Exception as e:
print(f"Unknown未知错误: {e}")
return False
def main():
producer = create_producer()
if not producer:
return
count = 0
try:
while True:
# 等待2秒,发送1条数据
data = generate_data()
success = send_data_kafka(producer, TOPIC_NAME, data)
if success:
count += 1
time.sleep(2)
except KeyboardInterrupt:
print(f"\n 用户中断程序")
except Exception as e:
print(f" 程序异常: {e}")
finally:
if producer:
producer.flush()
producer.close()
print(f" 生产者已关闭")
print(f" 总共发送: {count} 条消息")
if __name__ == "__main__":
main()
- 第5步，运行脚本，模拟 Producer（生产者）持续向指定主题（示例 mytopic）发布消息
注：若执行 python3 kafka_write.py 报错 No module named 'kafka'，改成python3完整路径即可，如下：
root@localhost:~# /usr/local/python3/bin/python3.12 kafka_write.py
数据已发送 - 分区: 2, 偏移量: 0, 数据: {'timestamp': '2026-07-24T18:15:30.079058', 'message_id': 4744, 'temperature': 29.22, 'humidity': 57.44, 'status': 'normal', 'data': 'sample_data_8'}
数据已发送 - 分区: 1, 偏移量: 0, 数据: {'timestamp': '2026-07-24T18:15:32.201306', 'message_id': 6586, 'temperature': 29.46, 'humidity': 41.33, 'status': 'error', 'data': 'sample_data_99'}
数据已发送 - 分区: 1, 偏移量: 1, 数据: {'timestamp': '2026-07-24T18:15:34.212115', 'message_id': 4128, 'temperature': 31.28, 'humidity': 53.67, 'status': 'error', 'data': 'sample_data_9'}
数据已发送 - 分区: 1, 偏移量: 2, 数据: {'timestamp': '2026-07-24T18:15:36.222534', 'message_id': 5927, 'temperature': 20.61, 'humidity': 69.11, 'status': 'normal', 'data': 'sample_data_83'}
数据已发送 - 分区: 1, 偏移量: 3, 数据: {'timestamp': '2026-07-24T18:15:38.233415', 'message_id': 8883, 'temperature': 34.24, 'humidity': 50.98, 'status': 'error', 'data': 'sample_data_11'}
数据已发送 - 分区: 1, 偏移量: 4, 数据: {'timestamp': '2026-07-24T18:15:40.245118', 'message_id': 6340, 'temperature': 29.93, 'humidity': 66.88, 'status': 'warning', 'data': 'sample_data_69'}
数据已发送 - 分区: 1, 偏移量: 5, 数据: {'timestamp': '2026-07-24T18:15:42.253378', 'message_id': 4797, 'temperature': 33.86, 'humidity': 47.79, 'status': 'warning', 'data': 'sample_data_50'}
数据已发送 - 分区: 2, 偏移量: 1, 数据: {'timestamp': '2026-07-24T18:15:44.262279', 'message_id': 7959, 'temperature': 21.11, 'humidity': 79.54, 'status': 'warning', 'data': 'sample_data_48'}
数据已发送 - 分区: 1, 偏移量: 6, 数据: {'timestamp': '2026-07-24T18:15:46.266677', 'message_id': 1128, 'temperature': 26.53, 'humidity': 68.53, 'status': 'normal', 'data': 'sample_data_30'}
数据已发送 - 分区: 1, 偏移量: 7, 数据: {'timestamp': '2026-07-24T18:15:48.274835', 'message_id': 4690, 'temperature': 28.22, 'humidity': 49.82, 'status': 'error', 'data': 'sample_data_81'}
数据已发送 - 分区: 0, 偏移量: 0, 数据: {'timestamp': '2026-07-24T18:15:50.283559', 'message_id': 4806, 'temperature': 32.27, 'humidity': 50.64, 'status': 'normal', 'data': 'sample_data_55'}
数据已发送 - 分区: 0, 偏移量: 1, 数据: {'timestamp': '2026-07-24T18:15:52.292744', 'message_id': 9001, 'temperature': 30.72, 'humidity': 65.27, 'status': 'warning', 'data': 'sample_data_42'}
数据已发送 - 分区: 0, 偏移量: 2, 数据: {'timestamp': '2026-07-24T18:15:54.301527', 'message_id': 3430, 'temperature': 25.49, 'humidity': 55.03, 'status': 'warning', 'data': 'sample_data_21'}
数据已发送 - 分区: 1, 偏移量: 8, 数据: {'timestamp': '2026-07-24T18:15:56.310459', 'message_id': 3990, 'temperature': 24.78, 'humidity': 71.34, 'status': 'error', 'data': 'sample_data_20'}
- 第6步，利用 Python 每 1 分钟从 Kafka 消费数据，并按照时间间隔命名保存文件，将下面脚本保存为 kafka_read.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import time
import json
from datetime import datetime
from kafka import KafkaConsumer
from kafka.errors import KafkaError
import warnings
warnings.filterwarnings("ignore")
# Kafka配置
KAFKA_BOOTSTRAP_SERVERS = '192.168.31.101:9092'
TOPIC_NAME = 'mytopic'
CONSUMER_GROUP = 'mygroup'
OUTPUT_DIR = './kafka_data' # 数据存储目录
INTERVAL_MINUTES = 1 # 时间间隔,单位分钟
class KafkaDataWriter:
def __init__(self, bootstrap_servers, topic, group_id, output_dir):
self.bootstrap_servers = bootstrap_servers
self.topic = topic
self.group_id = group_id
self.output_dir = output_dir
self.consumer = None
self.interval_start_time = None
os.makedirs(output_dir, exist_ok=True)
self.current_file_path = self._get_file_path()
self.data_buffer = []
def _get_file_path(self):
timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
return os.path.join(self.output_dir, f"data_{timestamp}.json")
def _flush_buffer(self):
if not self.data_buffer:
return
self.current_file_path = self._get_file_path()
with open(self.current_file_path, 'w', encoding='utf-8') as f:
for data in self.data_buffer:
f.write(json.dumps(data, ensure_ascii=False) + '\n')
print(f"时间达到: {INTERVAL_MINUTES} 分钟, 数据写入: {self.current_file_path}, 共计: ({len(self.data_buffer)} 条)")
self.data_buffer = []
def connect(self):
self.consumer = KafkaConsumer(
self.topic,
bootstrap_servers=self.bootstrap_servers,
group_id=self.group_id,
auto_offset_reset='earliest', # 从最早的消息开始消费
enable_auto_commit=True,
auto_commit_interval_ms=5000, # 每5秒提交一次
value_deserializer=lambda v: json.loads(v.decode('utf-8')) if v else None,
key_deserializer=lambda v: v.decode('utf-8') if v else None,
consumer_timeout_ms=1000
)
self.interval_start_time = datetime.now()
return True
def consume_messages(self):
print("\n" + "="*70)
print(f"成功连接: {self.bootstrap_servers}")
print(f"订阅主题: {self.topic}")
print(f"消费者组: {self.group_id}")
print(f"切换间隔: {INTERVAL_MINUTES} 分钟")
print(f"注意事项: 开始消费 {self.topic}, 按 Ctrl+C 随时停止")
print("="*70 + "\n")
msg_count = 0
try:
while True:
elapsed_minutes = (datetime.now() - self.interval_start_time).total_seconds() / 60.0
if elapsed_minutes >= INTERVAL_MINUTES:
self._flush_buffer()
self.interval_start_time = datetime.now()
for msg in self.consumer:
msg_count += 1
self.data_buffer.append({
'timestamp': datetime.now().isoformat(),
'partition': msg.partition,
'offset': msg.offset,
'key': msg.key,
'value': msg.value,
})
# 每10条消息打印一次状态
if msg_count % 10 == 0:
print(f"已接收: {msg_count} 条消息, 当前缓冲区: {len(self.data_buffer)} 条")
except KeyboardInterrupt:
print("\n用户中断")
finally:
self._flush_buffer()
if self.consumer:
self.consumer.close()
print("程序结束")
def main():
writer = KafkaDataWriter(
bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
topic=TOPIC_NAME,
group_id=CONSUMER_GROUP,
output_dir=OUTPUT_DIR
)
writer.connect()
writer.consume_messages()
if __name__ == "__main__":
main()
- 第7步，运行脚本，模拟持续消费指定主题（示例 mytopic）的消息
root@localhost:~# /usr/local/python3/bin/python3.12 kafka_read.py
======================================================================
成功连接: 192.168.31.101:9092
订阅主题: mytopic
消费者组: mygroup
切换间隔: 1 分钟
注意事项: 开始消费 mytopic, 按 Ctrl+C 随时停止
======================================================================
已接收: 10 条消息, 当前缓冲区: 10 条
已接收: 20 条消息, 当前缓冲区: 20 条
已接收: 30 条消息, 当前缓冲区: 30 条
已接收: 40 条消息, 当前缓冲区: 40 条
时间达到: 1 分钟, 数据写入: ./kafka_data/data_2026-07-24_18-20-18.json, 共计: (47 条)
已接收: 50 条消息, 当前缓冲区: 3 条
已接收: 60 条消息, 当前缓冲区: 13 条
已接收: 70 条消息, 当前缓冲区: 23 条
时间达到: 1 分钟, 数据写入: ./kafka_data/data_2026-07-24_18-21-20.json, 共计: (31 条)
已接收: 80 条消息, 当前缓冲区: 2 条
已接收: 90 条消息, 当前缓冲区: 12 条
已接收: 100 条消息, 当前缓冲区: 22 条
- 第8步，查看特定消费组的延迟
#.查看特定消费组的延迟
root@localhost:~# /data/kafka_9092/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.31.101:9092 --describe --group mygroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
mygroup mytopic 0 35 35 0 kafka-python-3.0.8-5c8df861-d85d-402c-80fc-76ab4781aab3 /192.168.31.102 kafka-python-3.0.8
mygroup mytopic 1 50 50 0 kafka-python-3.0.8-5c8df861-d85d-402c-80fc-76ab4781aab3 /192.168.31.102 kafka-python-3.0.8
mygroup mytopic 2 51 51 0 kafka-python-3.0.8-5c8df861-d85d-402c-80fc-76ab4781aab3 /192.168.31.102 kafka-python-3.0.8