python-can中Notifier類的實現(xiàn)報文的實時分發(fā)

更新時間：2026年04月10日 10:25:30 作者：名字太俊不提也罷

文章介紹了利用python-can庫中的Notifier類創(chuàng)建一個線程來從總線讀取CAN消息,并通過回調(diào)函數(shù)將消息分發(fā)給監(jiān)聽對象,具有一定的參考價值,感興趣的可以了解一下

前言

這個問題之前很早之前就解決了，今天記錄一下這個相關(guān)知識點，防止后續(xù)遺忘

一、Notifier是什么類

Notifier 是 python-can 庫中的消息分發(fā)器，它創(chuàng)建后臺線程從 CAN 總線持續(xù)讀取消息，并實時分發(fā)給所有注冊的監(jiān)聽器（Listeners）

Notifier 對象用作總線的消息分發(fā)器。Notifier 創(chuàng)建一個線程來從總線讀取消息并將它們分發(fā)給需要監(jiān)聽的對象

核心工作流程

CAN 總線 → Notifier（后臺線程） → Listener1 → 處理邏輯
                              → Listener2 → 日志記錄
                              → Listener3 → 隊列緩存

二、具體實現(xiàn)代碼

代碼如下（示例）：

class CanTool():
    def __init__(self):
        self.__mq = queue.Queue(50)
        self.__bus = None
        self.__notifier = None

    def notifier_init(self):
        """
        功能：初始化報文消息得回調(diào)函數(shù)
        """
        logger = can.Logger("logfile.asc")
        listeners = [
            self.put_mq,  # 回調(diào)函數(shù)，通過隊列返回報文消息
            logger,  # 保存log日志，asc格式
        ]
        self.__notifier = can.Notifier(self.__bus, listeners) #接收來自self.__bus的報文，分發(fā)給listeners對象


    def bus_init(self, device: Optional[str], channel:Optional[str], bitrate:Optional[int]):
        """
        功能：初始化總線設(shè)備
        param1：設(shè)備名。str
        param2：通道，str
        param3：波特率，int
        """
        try:
            self.__bus = can.interface.Bus(bustype=device, channel=channel, bitrate=bitrate)
            self.notifier_init()
            return self.__bus
        except Exception:
            return self.__bus


    def put_mq(self,msg: Optional[can.Message]):
        """
        功能：接收報文得回調(diào)函數(shù)
        param1：分發(fā)得到的報文對象
        """
        if not self.__mq.full():
            self.__mq.put_nowait(msg)	#通過隊列進行傳輸

    def busclose(self):
        """
        功能：結(jié)束總線收發(fā)，回收資源
        """
        self.__notifier.stop()	#停止分發(fā)
        self.__bus.shutdown()	#停止這個bus

總結(jié)

由于Python語言本身的問題，當程序讀取寫較大的總線消息時，會存在各種各樣的問題，本文中的方案，結(jié)合前面的報文篩選功能，能解決大部分這相關(guān)的問題。

重要注意事項

要點	說明
必須調(diào)用 stop()	否則監(jiān)聽器可能無法刷新緩沖區(qū)，導(dǎo)致數(shù)據(jù)丟失
線程安全	自定義監(jiān)聽器的 `on_message_received` 需考慮線程安全
避免阻塞	監(jiān)聽器中的處理應(yīng)盡量快速，耗時操作請用隊列
異常處理	監(jiān)聽器可實現(xiàn) `on_error()` 來捕獲異常