如何实现python-can的BLFWriter与Notifier实时CAN消息并行处理？

针对你的场景，最优方案是使用线程池（concurrent.futures.ThreadPoolExecutor）替代单线程创建，同时添加消息去重逻辑——因为你需要的是最新传输值，旧消息的解码结果没有意义，完全可以丢弃，避免不必要的计算。

核心思路 #

1. 线程池复用线程：限制线程数量（比如4-8个），避免10ms一次消息就创建新线程，大幅降低资源占用。
2. BLF写入优先：保留Notifier直接调用logger.on_message_received，这个操作本身轻量，不会阻塞，确保实时记录。
3. 消息去重：对每个CAN ID只保留最新的待解码消息，如果同一ID的旧解码任务还在排队，直接替换成新消息，避免无效解码。
4. 不阻塞GUI：线程池的任务在后台执行，customtkinter的主循环（事件循环）不受影响，解码完成后通过线程安全的方式（如after()方法）更新GUI即可。

优化后的代码示例 #

from can import Notifier, BLFWriter
from can.interfaces.pcan import PcanBus
from can.interfaces.vector import VectorBus
import time
from concurrent.futures import ThreadPoolExecutor
import threading

# 存储每个CAN ID的最新消息，线程锁保证操作安全
latest_messages = {}
msg_lock = threading.Lock()

# 线程池，限制最大线程数避免资源浪费
executor = ThreadPoolExecutor(max_workers=4)

def decode_message(msg):
    # 模拟解码耗时
    print(f"Decoding message ID: {hex(msg.arbitration_id)}")
    time.sleep(1)
    
    # 解码完成后更新字典（线程安全）
    with msg_lock:
        latest_messages[msg.arbitration_id] = msg.data  # 替换为你的实际解析结果
    
    # 如果需要更新GUI，必须回到主线程执行
    # app.after(0, update_gui, msg.arbitration_id, parsed_result)

def handle_decode(msg):
    # 标记当前ID的消息为待解码，替换旧的未处理消息
    with msg_lock:
        latest_messages[msg.arbitration_id] = "PENDING"
    
    # 提交解码任务到线程池
    executor.submit(process_decode_task, msg)

def process_decode_task(msg):
    # 检查当前消息是否仍为最新版本，避免解码已过期的消息
    with msg_lock:
        if latest_messages.get(msg.arbitration_id) != "PENDING":
            return  # 不是最新消息，直接跳过
    
    # 执行解码逻辑
    decode_message(msg)

def main():
    file = "testing.blf"
    logger = BLFWriter(file=file)

    can_bus = PcanBus(interface="pcan", channel="PCAN_USBBUS1", bitrate=500000, device_id=0xFF, receive_own_messages=True)
    # can_bus = VectorBus(interface="vector", channel=0, bitrate=500000, serial=96869, receive_own_messages=True)

    # Notifier绑定BLF写入和自定义解码处理函数
    notifier = Notifier(can_bus, [logger.on_message_received, handle_decode])

    try:
        print("Running... Press Ctrl+C to stop.")
        while True:
            time.sleep(1)  # GUI场景下替换为customtkinter的主循环（app.mainloop()）
    except KeyboardInterrupt:
        pass
    finally:
        notifier.stop()
        can_bus.shutdown()
        logger.stop()
        executor.shutdown(wait=True)  # 等待所有解码任务完成后关闭线程池

if __name__ == "__main__":
    main()

关键细节说明 #

• 线程池大小：根据CPU核心数调整max_workers，4-8个线程足以应对10ms频率的消息，同时避免资源浪费。
• 消息去重逻辑：通过latest_messages字典标记待解码状态，确保只有最新的消息被解码，避免旧消息占用线程资源。
• GUI线程安全：customtkinter的GUI组件必须在主线程更新，解码完成后要用app.after(0, callback, args)将更新操作抛回主线程，规避线程安全问题。
• 为什么不用multiprocessing？：多进程的进程间通信开销远大于线程，若你的解码任务不是纯CPU密集型（如大量复杂计算），线程池效率更高、代码更简洁。
• 为什么不用asyncio？：python-can的Notifier基于线程实现，和asyncio混合需要额外适配，且你的解码任务是同步阻塞型，asyncio优势不明显，反而不如线程池直接。

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