如何将Kafka与速度传感器对接并通过Kafka-Python实现车速数据采集存储

整体链路架构梳理 #

完整的实现链路为：速度传感器 → 边缘数据采集端 → Kafka生产者 → Kafka集群 → Kafka消费者 → 业务数据库。核心对接逻辑集中在边缘采集端的协议转换和生产者配置上。

生产者与速度传感器的对接方案选型 #

根据现有传感器的输出协议，直接选对应的适配方案即可：

• 传感器支持HTTP/MQTT主动上报：适用于分散部署、具备联网能力的智能传感器，传感器直接把数据推送到边缘端的接收接口，接口侧调用kafka-python生产者写入Kafka即可，对接成本最低。
• 传感器通过串口/Modbus等工业协议输出：适用于固定路段集中部署的工业级传感器，边缘端先通过硬件接口读取传感器的原始输出，完成协议解析后再调用生产者写入Kafka，传输稳定性更高。

具体实现步骤 #

1. 前置环境准备 #

• 提前部署好Kafka集群，测试场景用单节点即可，生产环境建议至少3节点，提前创建名为vehicle_speed的Topic，传感器接入量100台以内的话设置3分区、2副本即可。
• 安装基础依赖：pip install kafka-python，如果是串口/Modbus协议的传感器，额外安装pip install pyserial pymodbus。

2. 核心代码实现 #

HTTP上报场景示例

可配合FastAPI/Flask搭建接收接口，核心生产者逻辑如下：

from kafka import KafkaProducer
import json

# 初始化生产者，参数根据业务要求调整
producer = KafkaProducer(
    bootstrap_servers=['你的Kafka节点IP:9092'],
    # 数据序列化为JSON，数据量大的场景可替换为Avro/Protobuf压缩体积
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    # 可靠性要求一般选acks=1，要求完全不丢数据选acks=all
    acks=1,
    retries=3
)

# 传感器上报数据处理逻辑
def handle_sensor_upload(receive_data):
    # 前置数据校验，过滤脏数据
    if not all(k in receive_data for k in ["sensor_id", "speed", "timestamp"]):
        return
    # 按传感器ID做分区路由，保证同一传感器的数据顺序一致
    producer.send(
        topic="vehicle_speed",
        value=receive_data,
        key=str(receive_data["sensor_id"]).encode("utf-8")
    )

# 测试调用
if __name__ == "__main__":
    test_data = {
        "sensor_id": "road_003_sensor_017",
        "plate_number": "京A88990",
        "speed": 58,
        "timestamp": 1701234567000
    }
    handle_sensor_upload(test_data)
    producer.flush()

串口协议传感器示例

import serial
from kafka import KafkaProducer
import json

# 初始化串口连接，参数按传感器说明调整
ser = serial.Serial("/dev/ttyUSB0", 9600, timeout=1)
producer = KafkaProducer(
    bootstrap_servers=['你的Kafka节点IP:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks=1,
    retries=3
)

while True:
    raw_data = ser.readline().decode("utf-8").strip()
    if not raw_data:
        continue
    # 按传感器输出协议解析，示例格式为 传感器ID,车牌号,速度,时间戳
    sensor_id, plate, speed, ts = raw_data.split(",")
    formatted_data = {
        "sensor_id": sensor_id,
        "plate_number": plate,
        "speed": int(speed),
        "timestamp": int(ts)
    }
    producer.send(
        topic="vehicle_speed",
        value=formatted_data,
        key=sensor_id.encode("utf-8")
    )

配置优化建议 #

• 接入传感器数量超过200台的话，可以给生产者配置linger_ms=5，批量攒消息发送，降低集群压力，吞吐量能提升30%以上。
• 边缘端网络不稳定的场景，可开启生产者本地缓冲，无需额外做持久化逻辑，kafka-python自带的重试和缓冲机制就能避免短时间网络中断导致的数据丢失。
• 后续消费者写入数据库时，建议批量攒100~500条数据再做批量写入，性能比单条插入高10倍以上。

内容的提问来源于stack exchange，提问作者Db2Cramp