构建可靠的串行通信抽象层：解决多线程并发问题

心靈之曲

发布时间：2025-07-03 22:04:01

1143人浏览过

来源于php中文网

原创

构建可靠的串行通信抽象层：解决多线程并发问题

在多线程环境中，对串行通信设备进行并发访问常面临通信冲突和协议违背的挑战。本文旨在探讨如何构建一个高层抽象来解决这些问题。文章详细介绍了两种核心策略：一是通过设立专用串行通信处理线程，利用消息队列实现请求的序列化处理；二则是运用互斥锁（Mutex）机制，确保对串口的独占访问。这些方法能够有效管理并发请求，保障数据完整性与通信协议的正确执行，从而实现简洁且可靠的多线程串口操作。

串行通信的并发挑战

当多个线程需要与同一个串行设备进行通信时，直接的、无同步的访问会导致严重问题。例如，一个线程可能需要持续查询设备状态（如温度），而另一个线程则可能在随机时间点发送控制命令。如果两个线程同时尝试写入串口或读取数据，看似会发生“数据混淆”，但实际上，底层操作系统驱动程序通常会避免字节级别的交错。真正的核心问题在于：

协议违背： 大多数串行设备，特别是采用主从模式的设备，被设计为一次只处理一个请求并返回一个响应。如果在一个请求-响应周期完成之前，另一个线程就发送了新的请求，设备可能会进入不确定状态，导致数据丢失或通信错误。
状态管理： 缺乏统一的协调机制，各个线程无法感知其他线程的通信状态，从而无法保证通信的顺序性和完整性。

因此，为了确保通信的可靠性，我们必须在应用程序层面实现高级抽象，来管理和同步对串行端口的访问。

策略一：专用串行通信处理线程

一种强大且优雅的解决方案是引入一个专用的串行通信处理线程。该线程作为所有串行I/O操作的唯一协调者和执行者。其他需要与串行设备通信的线程，不再直接操作串口，而是将它们的请求发送到这个专用线程的消息队列中。

工作原理：

请求队列： 设立一个线程安全的队列（如Python的queue.Queue），用于存储来自其他线程的串行通信请求。每个请求通常包含要发送的数据、期望的响应长度以及一个用于通知请求线程响应已到达的机制（例如，一个事件对象或另一个回调队列）。
单一入口： 专用线程持续地从队列中取出请求。
序列化执行： 对于每个取出的请求，专用线程负责：
- 向串行端口写入请求数据。
- 等待并读取设备的响应数据（通常是阻塞式读取，直到收到完整响应或超时）。
- 将响应数据（或错误信息）发送回发起请求的线程。
响应回传： 请求线程在将请求放入队列后，会进入等待状态（例如，等待一个事件被设置，或者从一个特定的响应队列中接收数据），直到专用线程处理完其请求并返回结果。

示例概念：

import threading
import queue
import serial
import time

class SerialDeviceAbstraction:
    def __init__(self, port, baudrate):
        self.serial_port = serial.Serial(port, baudrate, timeout=1) # timeout for read
        self.request_queue = queue.Queue()
        self.response_map = {} # To map request IDs to response queues/events
        self.handler_thread = threading.Thread(target=self._serial_handler_loop, daemon=True)
        self.handler_thread.start()
        self.request_id_counter = 0
        self.lock = threading.Lock() # For generating unique request IDs

    def _get_next_request_id(self):
        with self.lock:
            self.request_id_counter += 1
            return self.request_id_counter

    def _serial_handler_loop(self):
        while True:
            # Wait for a request
            request_data, response_event, request_id = self.request_queue.get()

            try:
                # 1. Write request
                self.serial_port.write(request_data)
                # Ensure all data is sent before reading (optional, depends on hardware)
                # self.serial_port.flush() 

                # 2. Read response (blocking read with timeout)
                # This assumes a fixed response length or a clear end-of-message delimiter
                response = self.serial_port.read(8) # Example: read 8 bytes

                # 3. Store response and notify original thread
                self.response_map[request_id] = response
                response_event.set() # Signal that response is ready

            except serial.SerialException as e:
                print(f"Serial communication error: {e}")
                self.response_map[request_id] = None # Indicate error
                response_event.set()
            finally:
                self.request_queue.task_done() # Mark task as done

    def get(self, query_bytes):
        request_id = self._get_next_request_id()
        response_event = threading.Event()

        # Enqueue the request
        self.request_queue.put((query_bytes, response_event, request_id))

        # Wait for the response
        response_event.wait() # Blocks until the handler thread signals

        # Retrieve the response
        response = self.response_map.pop(request_id, None)
        if response is None:
            raise IOError("Failed to get response from serial device.")
        return response

# Usage example (conceptual)
# serial_device_abstraction = SerialDeviceAbstraction(port="/dev/ttyUSB0", baudrate=9600)

# def thread1():
#     while True:
#         try:
#             data = serial_device_abstraction.get(b"foo_query")
#             print(f"Thread 1 received: {data}")
#         except IOError as e:
#             print(f"Thread 1 error: {e}")
#         time.sleep(1)

# def thread2():
#     time.sleep(random.random())
#     try:
#         data = serial_device_abstraction.get(b"bar_query")
#         print(f"Thread 2 received: {data}")
#     except IOError as e:
#         print(f"Thread 2 error: {e}")

# threading.Thread(target=thread1).start()
# threading.Thread(target=thread2).start()

优点：

强封装性： 将所有底层串口操作和并发处理逻辑封装在一个专用线程中，其他线程无需关心细节。
自然序列化： 请求在队列中排队，由专用线程顺序执行，天然地解决了协议违背问题。
简化客户端： 客户端线程的代码变得非常简洁，只需调用高级抽象接口。

策略二：基于互斥锁的独占访问

另一种实现高层抽象的方法是使用互斥锁（Mutex）来强制对串行端口的独占访问。这种方法不依赖于一个专用的处理线程，而是让每个需要访问串口的线程在进行I/O操作前，先获取互斥锁。

TapNow

新一代AI视觉创作引擎

下载

工作原理：

共享资源： 串行端口的文件描述符（或其封装对象）和互斥锁被视为共享资源。
临界区： 所有对串行端口的写入和读取操作都被定义为“临界区”。
获取锁： 任何线程在进入临界区之前，必须首先尝试获取互斥锁。如果锁已被其他线程持有，当前线程将被阻塞，直到锁被释放。
执行I/O： 成功获取锁的线程可以安全地执行串行I/O操作（写入请求，然后读取响应）。
释放锁： I/O操作完成后，线程必须立即释放互斥锁，以便其他等待的线程可以继续执行。

示例伪代码：

import threading
import serial
import time

# 假设 serial_port 是全局或类实例的串行端口对象
# 假设 serial_lock 是全局或类实例的互斥锁对象

class SerialDeviceAbstractionMutex:
    def __init__(self, port, baudrate):
        self.serial_port = serial.Serial(port, baudrate, timeout=1)
        self.serial_lock = threading.Lock()

    def send_receive(self, request_msg_bytes, response_len):
        """
        通过串行端口发送请求并接收响应。
        所有对串口的读写操作都通过互斥锁保护。
        """
        response_data = None
        with self.serial_lock: # 自动获取锁并在退出with块时释放锁
            try:
                # 1. 写入请求
                self.serial_port.write(request_msg_bytes)
                # 确保所有数据已发送 (对于某些驱动可能不需要，但有助于确保时序)
                # self.serial_port.flush() 

                # 2. 读取响应 (阻塞模式等待)
                response_data = self.serial_port.read(response_len)
                if len(response_data) < response_len:
                    # 处理响应不完整的情况，可能需要更复杂的协议解析
                    raise IOError(f"Incomplete response received: expected {response_len}, got {len(response_data)}")

            except serial.SerialException as e:
                print(f"Serial communication error: {e}")
                raise # 重新抛出异常，让调用者处理
            except Exception as e:
                print(f"An unexpected error occurred: {e}")
                raise

        return response_data

# Usage example (conceptual)
# serial_device_abstraction_mutex = SerialDeviceAbstractionMutex(port="/dev/ttyUSB0", baudrate=9600)

# def thread1_mutex():
#     while True:
#         try:
#             data = serial_device_abstraction_mutex.send_receive(b"foo_query", 8)
#             print(f"Thread 1 (Mutex) received: {data}")
#         except IOError as e:
#             print(f"Thread 1 (Mutex) error: {e}")
#         time.sleep(1)

# def thread2_mutex():
#     time.sleep(random.random())
#     try:
#         data = serial_device_abstraction_mutex.send_receive(b"bar_query", 8)
#         print(f"Thread 2 (Mutex) received: {data}")
#     except IOError as e:
#         print(f"Thread 2 (Mutex) error: {e}")

# threading.Thread(target=thread1_mutex).start()
# threading.Thread(target=thread2_mutex).start()

优点：

实现相对简单： 对于简单的请求-响应模式，只需在I/O操作外部添加锁机制即可。
直接控制： 每个线程直接控制其何时访问串口。

注意事项：

死锁风险： 如果锁的获取和释放逻辑处理不当，可能导致死锁。使用with语句管理锁（如Python的threading.Lock）可以有效避免忘记释放锁的问题。
超时处理： read操作需要适当的超时设置，以防设备无响应导致线程永久阻塞。

关键考量与最佳实践

无论选择哪种策略，以下几点是构建可靠串行通信抽象时必须考虑的：

严格遵循请求-响应协议： 确保在发送下一个请求之前，前一个请求的完整响应（或超时）已经被处理。这是避免设备状态混乱的关键。
错误处理和超时机制： 串行通信容易受到物理干扰或设备无响应的影响。必须实现健壮的错误检测（如校验和）和超时机制。当通信失败时，应有明确的错误报告和恢复策略。
缓冲与流控制： 考虑串口的输入/输出缓冲区大小。在高速通信中，可能需要额外的软件缓冲和流控制机制来防止数据溢出。
可重入性与线程安全： 确保你的抽象层是线程安全的。所有共享资源（如串口对象、队列、锁）都必须正确同步。
选择合适的策略：
- 如果通信模式复杂，需要复杂的请求调度、优先级处理或与设备保持长期会话，专用串行通信处理线程通常是更优的选择，因为它提供了更强大的控制和更清晰的逻辑分离。
- 如果通信模式简单，主要是短促的请求-响应，且对实时性要求不是极高，基于互斥锁的独占访问可能更易于实现和维护。

总结

为多线程环境下的串行通信构建高层抽象是确保系统稳定性和可靠性的关键。通过采用专用串行通信处理线程或基于互斥锁的独占访问机制，我们可以有效地解决并发访问带来的通信冲突和协议违背问题。这两种策略各有优势，开发者应根据具体的应用场景和需求，选择最适合的方案，并结合完善的错误处理和协议管理，以构建健壮、高效的串行通信系统。最终目标是让上层应用线程能够以一种简洁、无需感知底层并发细节的方式，安全地与串行设备进行交互。

Python 如何让 subprocess 捕获彩色输出（ANSI escape）

Python 如何让 dict 按照插入顺序遍历（OrderedDict vs 3.7+）

Python pandas 如何让 read_csv 自动识别哪些列是日期

Python 如何让一个函数记住它的调用次数和历史参数

Python 如何判断一个文件是否正在被其他进程写入