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

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()

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()