高并发环境下串口通信的高级抽象与实现

1. 串口通信中的并发挑战

在硬件设备通信场景中，尤其涉及到串口（serial port）通信时，经常会遇到多线程并发访问的需求。例如，一个线程可能需要持续地查询设备状态（如温度），而另一个线程可能在随机时间点发送一次性控制命令。如果不对串口访问进行有效管理，直接在多个线程中调用串口读写操作，将导致以下问题：

• 请求冲突与数据损坏（逻辑层面）：串口通信通常遵循严格的请求-响应（Request-Response）协议。远程设备在接收到请求后，会进入处理状态直到发送响应。如果主机端在未收到前一个请求的响应之前，就发送了新的请求，远程设备可能无法正确处理，导致数据错乱或通信失败。这并非因为比特或字节级别的混淆，而是因为违反了协议的时序要求。
• 缺乏同步机制：操作系统内核驱动程序处理I/O操作，线程本身无法直接控制硬件。并发问题并非源于比特流混淆，而是因为多个线程同时尝试发起I/O系统调用，而没有机制来确保这些操作的原子性和顺序性，从而无法保证请求-响应周期的完整性。

因此，构建一个高级抽象层来管理串口通信，并自动处理底层的并发问题，是实现稳定、可靠通信的关键。

2. 解决方案一：基于队列的独立通信线程

一种优雅且推荐的解决方案是引入一个专门的线程来负责所有的串口I/O操作。这个线程充当串口通信的“管家”，所有其他线程的串口请求都通过一个队列提交给它。

2.1 工作原理

1. 单一入口点：创建一个独立的线程，该线程拥有对串口的独占访问权。所有对串口的读写操作都只能通过这个线程进行。
2. 请求队列：其他需要与串口通信的线程（如查询温度的线程、发送控制命令的线程）不再直接操作串口，而是将它们的请求封装成消息，连同期望接收响应的地址（例如一个回调函数或另一个队列），放入一个共享的请求队列中。
3. 顺序处理：串口通信线程不断地从请求队列中取出请求。对于每个请求，它会执行以下步骤：
   • 向串口发送请求消息。
   • 等待并接收来自设备的响应（通常通过阻塞式读取实现）。
   • 将接收到的响应（或超时错误）发送回发起请求的线程（通过之前提供的地址）。
4. 抽象并发：通过这种方式，所有线程对串口的竞争都被隐藏在队列机制之后。串口通信被强制序列化，确保了请求-响应协议的严格遵守。

2.2 概念伪代码

import queue
import threading
import time
import random

# 假设的串口操作函数
def _serial_write(data):
    print(f"[Serial] Sending: {data}")
    time.sleep(0.1) # 模拟写入延迟

def _serial_read(size):
    # 模拟读取响应
    response_map = {
        "foo": "temp_data",
        "bar": "config_ack"
    }
    # 假设响应与请求直接相关，实际可能需要解析设备返回
    # 这里简化为根据发送的数据模拟返回
    if "foo" in threading.current_thread().name: # 模拟foo请求的响应
        return response_map["foo"]
    elif "bar" in threading.current_thread().name: # 模拟bar请求的响应
        return response_map["bar"]
    return "unknown_response"

class SerialCommunicator:
    def __init__(self):
        self.request_queue = queue.Queue()
        self.response_queues = {} # 存储每个请求对应的响应队列
        self.serial_thread = threading.Thread(target=self._run_serial_handler, name="SerialHandlerThread")
        self.serial_thread.daemon = True
        self.serial_thread.start()

    def _run_serial_handler(self):
        while True:
            request_item = self.request_queue.get()
            query = request_item['query']
            response_queue = request_item['response_queue']

            try:
                # 严格的请求-响应周期
                _serial_write(query)
                response = _serial_read(8) # 假设读取8字节
                response_queue.put({"status": "success", "data": response})
            except Exception as e:
                response_queue.put({"status": "error", "message": str(e)})
            finally:
                self.request_queue.task_done()

    def send_query(self, query):
        # 每个请求创建一个临时的响应队列
        response_q = queue.Queue(1)
        self.request_queue.put({"query": query, "response_queue": response_q})
        # 阻塞等待响应
        result = response_q.get()
        if result['status'] == 'success':
            print(f"[{threading.current_thread().name}] Received response for '{query}': {result['data']}")
            return result['data']
        else:
            print(f"[{threading.current_thread().name}] Error for '{query}': {result['message']}")
            raise Exception(result['message'])

# 实例化串口通信抽象层
serial_device_abstraction = SerialCommunicator()

# 示例：线程1持续查询"foo"
def thread1_task():
    threading.current_thread().name = "Thread-Foo"
    while True:
        try:
            serial_device_abstraction.send_query("foo")
        except Exception as e:
            print(f"Thread-Foo encountered error: {e}")
        time.sleep(1)

# 示例：线程2随机查询"bar"
def thread2_task():
    threading.current_thread().name = "Thread-Bar"
    for _ in range(5): # 模拟查询5次
        time.sleep(random.uniform(0.1, 2.0))
        try:
            serial_device_abstraction.send_query("bar")
        except Exception as e:
            print(f"Thread-Bar encountered error: {e}")

# 启动线程
t1 = threading.Thread(target=thread1_task)
t2 = threading.Thread(target=thread2_task)

t1.start()
t2.start()

# 等待一段时间，观察输出
time.sleep(10)
print("Main thread exiting.")

3. 解决方案二：基于互斥锁（Mutex）的独占访问

另一种相对直接的方案是使用互斥锁（Mutex）来保护串口的临界区。所有需要访问串口的线程在进行读写操作前，都必须先获取这个互斥锁。

绘蛙AI商品图

电商场景的AI创作平台，无需高薪聘请商拍和文案团队，使用绘蛙即可低成本、批量创作优质的商拍图、种草文案

下载

3.1 工作原理

1. 共享资源与互斥锁：将串口的文件描述符（或其封装对象）和互斥锁定义为全局或可访问的共享资源。
2. 临界区保护：任何线程在执行串口的写入和读取操作之前，必须先尝试获取互斥锁。如果锁已被其他线程持有，当前线程将被阻塞，直到锁被释放。
3. 释放锁：完成串口的读写操作后，线程必须立即释放互斥锁，以便其他等待的线程可以继续执行。
4. 强制协议：关键在于，获取锁后不仅要写入，还要等待并读取响应，确保一个完整的请求-响应周期完成后再释放锁。

3.2 伪代码示例

// 假设 serial_fd 是串口的文件描述符
// 假设 serial_mutex 是一个全局或静态的互斥锁
// (在Python中，可以使用threading.Lock)

// C/C++ 伪代码示例
procedure serial_messaging(u8 *request_mesg, int rqlen, u8 *response_mesg, int rslen)
{
    int rc;

    acquire_the_mutex(serial_mutex);    /* 获取互斥锁，否则当前线程阻塞 */

    rc = write(serial_fd, request_mesg, rqlen);
    if (rc < 0) {
        // 处理写入错误
        handle_error_condition();
        release_the_mutex(serial_mutex);
        return;
    }
    // tcdrain(serial_fd); // 对于一些系统，可能需要确保所有数据已发送，再计时或读取

    rc = read(serial_fd, response_mesg, rslen);   /* 使用阻塞模式等待响应 */
    if (rc < 0) {
        // 处理读取错误或超时
        handle_error_condition();
        release_the_mutex(serial_mutex);
        return;
    }

    release_the_mutex(serial_mutex);    /* 释放互斥锁，允许其他线程使用串口 */
    return;   /* 返回接收到的数据在 response_mesg 中 */
}

// 示例调用 (在不同的线程中)
void thread_foo_task() {
    u8 request[] = "foo";
    u8 response[8];
    while (true) {
        serial_messaging(request, sizeof(request), response, sizeof(response));
        // 处理接收到的 response
        sleep(1);
    }
}

void thread_bar_task() {
    u8 request[] = "bar";
    u8 response[8];
    sleep(random_delay());
    serial_messaging(request, sizeof(request), response, sizeof(response));
    // 处理接收到的 response
}

4. 注意事项与总结

• 严格遵守请求-响应协议：无论是哪种抽象方式，核心都是确保在发送下一个请求之前，前一个请求的响应已经被完全接收或已超时。这是避免通信混乱的关键。
• 错误处理和超时机制：在实际应用中，必须加入健壮的错误处理和超时机制。串口通信可能因为设备无响应、数据损坏或物理连接问题而失败。超时机制能防止线程无限期地等待响应。
• 选择合适的抽象：
  • 基于队列的独立线程：适用于复杂的、高并发的系统，能够提供更高级的抽象，将串口通信的细节完全封装。它天然地实现了请求的序列化，降低了其他线程的复杂性。
  • 基于互斥锁：适用于相对简单的场景，或者当开发者希望更直接地控制每个请求的生命周期时。它要求每个调用者都明确地管理锁的获取和释放，但如果实现得当，同样能确保正确性。
• 阻塞与非阻塞I/O：上述示例都倾向于使用阻塞式I/O，这简化了等待响应的逻辑。但在某些高性能或事件驱动的系统中，可能需要结合非阻塞I/O和事件循环来处理串口数据。

通过采用上述任一高级抽象方案，开发者可以有效地解决多线程环境下串口通信的并发问题，构建出稳定、高效且易于维护的设备通信模块。