Django/Python串行线并发问题：嵌入式Linux网关开发技术问询

嘿，刚好有过类似的串口采集+Django服务的落地经验，给你梳理几个解决串行线并发问题的核心方向，都是实际项目里验证过的：

RS485/Modbus串口是独占式硬件资源——同一时间只能有一个读写操作，否则会直接导致Modbus帧冲突、数据乱码甚至设备通信异常。你的前端AJAX循环请求如果同时触发多个Django视图去访问串口，必然会踩这个坑，这就是你要解决的核心问题。

1. 独立串口守护进程（最推荐，生产环境首选） #

把串口采集逻辑从Django视图里彻底抽离，做成一个后台独立守护进程：

• 这个进程独占串口，持续/定时执行Modbus采集，把结果存在共享存储（Redis、本地数据库、甚至跨进程共享内存）
• Django视图只负责从共享存储里读取数据返回给前端，完全不碰串口
• 优势：彻底规避并发冲突，性能拉满，Django可以轻松处理大量AJAX请求；采集逻辑和Web服务解耦，单独调试、维护都方便

举个伪代码示例：

# 串口采集守护进程（可以用systemd做成系统服务）
import serial
import modbus_tk.defines as cst
from modbus_tk import modbus_rtu
import redis
import time

# 初始化Redis作为共享存储
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# 初始化Modbus RTU客户端
serial_port = serial.Serial(
    '/dev/ttyRS485', baudrate=9600, parity='N', 
    stopbits=1, bytesize=8, timeout=1
)
modbus_master = modbus_rtu.RtuMaster(serial_port)
modbus_master.set_timeout(2.0)

while True:
    try:
        # 读取电气设备的保持寄存器（示例：地址1的设备，读取0-10寄存器）
        register_data = modbus_master.execute(1, cst.READ_HOLDING_REGISTERS, 0, 10)
        # 把数据存入Redis哈希结构，方便Django读取
        for idx, val in enumerate(register_data):
            redis_client.hset("electrical_device", f"register_{idx}", val)
        time.sleep(2)  # 采集间隔，按需调整
    except Exception as e:
        print(f"采集异常: {str(e)}")
        time.sleep(5)  # 出错后延迟重试

对应的Django视图就很简单了：

# Django视图示例
from django.http import JsonResponse
import redis

def get_device_data(request):
    redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
    # 从Redis读取最新采集的数据
    device_data = redis_client.hgetall("electrical_device")
    # 格式化为前端需要的结构
    formatted_data = {k: int(v) for k, v in device_data.items()}
    return JsonResponse({"status": "success", "data": formatted_data})

2. Django视图内的跨进程锁机制（快速临时方案） #

如果不想单独做守护进程，也可以用锁来保证同一时间只有一个请求访问串口：

• 注意：如果Django是多进程部署（比如uWSGI多worker、Gunicorn多进程），普通的threading.Lock没用，必须用跨进程锁（比如Redis分布式锁）
• 劣势：频繁的AJAX请求会导致很多请求被拒绝（返回忙状态），用户体验一般；而且每次请求都要打开/关闭串口，效率比守护进程低很多

示例代码（Redis分布式锁）：

from django.http import JsonResponse
import redis
import serial
import modbus_tk.defines as cst
from modbus_tk import modbus_rtu

redis_client = redis.Redis(host='localhost', port=6379, db=0)
LOCK_KEY = "serial_port_exclusive_lock"
LOCK_TIMEOUT = 5  # 锁超时，防止死锁

def get_device_data(request):
    # 尝试获取分布式锁
    lock_acquired = redis_client.set(LOCK_KEY, "locked", ex=LOCK_TIMEOUT, nx=True)
    if not lock_acquired:
        return JsonResponse(
            {"status": "busy", "message": "正在采集数据，请稍后重试"},
            status=429
        )
    
    try:
        # 初始化串口和Modbus客户端
        serial_port = serial.Serial(
            '/dev/ttyRS485', baudrate=9600, parity='N', 
            stopbits=1, bytesize=8, timeout=1
        )
        modbus_master = modbus_rtu.RtuMaster(serial_port)
        modbus_master.set_timeout(2.0)
        # 采集数据
        register_data = modbus_master.execute(1, cst.READ_HOLDING_REGISTERS, 0, 10)
        formatted_data = {f"register_{i}": val for i, val in enumerate(register_data)}
        return JsonResponse({"status": "success", "data": formatted_data})
    except Exception as e:
        return JsonResponse({"status": "error", "message": str(e)}, status=500)
    finally:
        # 无论成功失败，都释放锁
        redis_client.delete(LOCK_KEY)

3. 额外优化建议 #

• Modbus库选型：推荐modbus-tk或者pymodbus，都是Python生态里成熟稳定的库，完美支持RTU模式
• 嵌入式Linux串口配置：记得给串口设备设置正确权限（比如sudo chmod 666 /dev/ttyRS485），或者把Django进程所属用户加入tty/dialout组，避免权限问题
• 前端通信优化：别用太频繁的AJAX轮询（比如1-5秒一次足够），可以换成WebSocket实现服务器主动推送数据——Django可以用Channels框架实现WebSocket，这样既能减少服务器压力，又能让前端实时获取数据

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