多进程替代多线程接收UART/USB数据是否合理？Pipe通信优化咨询

Hey there! Let's tackle your questions one by one and get that code working smoothly and elegantly.

1. Is using multiprocessing instead of multithreading for full parallelism reasonable? #

Your choice is totally valid—here’s why:
Since reading from UART and USB are IO-bound tasks (most of the time the process waits for data, not crunching numbers), both threads and processes work here. But multiprocessing gives you key advantages:

• If you later add CPU-heavy data processing (like parsing or transforming incoming data), you’ll avoid Python’s GIL (Global Interpreter Lock) limitation, which can bottleneck threads for CPU work.
• Each process has its own isolated memory space, so you don’t have to worry about thread-safety locks or race conditions if you expand your logic.
• If one communication task crashes (e.g., UART connection drops), it won’t take down the main process or USB reader—more robust for long-running data collection.

2. Is choosing Pipe over Queue a good call? #

Absolutely. Queue is built on top of Pipe but adds extra locking and serialization for scenarios where multiple processes need to read/write to the same queue. Since you’re doing simple one-way communication (workers send data only to the main process), Pipe is lighter and faster—your research about speed checks out.

3. Fixing the "only UART data prints" issue & making the code elegant #

Your original loop checks for UART data first, and its higher frequency can make USB data feel ignored (or buried in output). The solution is to stop polling and instead listen for incoming data on both pipes simultaneously using Python’s select module. This way, the main process reacts immediately whenever either pipe has data ready—no busy waiting, no missed USB packets.

Here’s the improved code with timestamp sync and clean handling:

import multiprocessing
from multiprocessing import Process, Pipe
import time
import select
# Import your serial/USB libraries here (example):
# import serial
# uart = serial.Serial('/dev/ttyUSB0', 9600)
# usb_device = ... # Initialize your USB device

def read_uart(uart_conn):
    while True:
        # Replace with actual UART read logic
        reading = uart.readline().strip()  # Remove newline/whitespace
        if reading:  # Only send non-empty data
            uart_conn.send(reading)

def read_usb(usb_conn):
    while True:
        # Replace with actual USB read logic
        reading = usb_device.readline().strip()
        if reading:
            usb_conn.send(reading)

if __name__ == '__main__':
    parent_usb_conn, usb_conn = Pipe()
    parent_uart_conn, uart_conn = Pipe()

    # Start worker processes
    process_usb = Process(target=read_usb, args=(usb_conn,))
    process_uart = Process(target=read_uart, args=(uart_conn,))
    process_usb.start()
    process_uart.start()

    # Map file descriptors to their source and connection
    fd_list = [parent_usb_conn.fileno(), parent_uart_conn.fileno()]
    conn_map = {
        parent_usb_conn.fileno(): ("USB", parent_usb_conn),
        parent_uart_conn.fileno(): ("UART", parent_uart_conn)
    }

    try:
        while True:
            # Wait for any pipe to have data ready
            ready, _, _ = select.select(fd_list, [], [], None)
            for fd in ready:
                source, conn = conn_map[fd]
                data = conn.recv()
                # Add millisecond-precision timestamp
                timestamp = time.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
                print(f"[{timestamp}] {source}: {data}")
    except KeyboardInterrupt:
        # Clean up gracefully on Ctrl+C
        process_usb.terminate()
        process_uart.terminate()
        process_usb.join()
        process_uart.join()
        # Close your hardware devices here
        # uart.close()
        # usb_device.close()
        print("\nProcesses terminated gracefully.")

Key improvements: #

• Uses select.select() to efficiently wait for incoming data, so the main process reacts instantly to new data from either source
• Adds millisecond-precision timestamps to sync data from both devices
• Includes a clean exit handler for manual interruption (Ctrl+C)
• Filters out empty read lines to avoid clutter
• Uses a mapping to track pipe sources, making the code scalable if you add more devices later

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