python blinker庫學習

legendmohe 發布于2019-07-25 10:33 / 1163人閱讀

摘要：的信號機制就是基于它建立的。觸發信號使用方法通知信號訂閱者。每個元組的組成為。與是兩個不同的信號。這時，可以使用優化信號發送信號通常會進行優化，以便快速的發送。

參考 Blinker Documentation

Blinker 是一個基于Python的強大的信號庫，它既支持簡單的對象到對象通信，也支持針對多個對象進行組播。Flask的信號機制就是基于它建立的。

Blinker的內核雖然小巧，但是功能卻非常強大，它支持以下特性：

支持注冊全局命名信號

支持匿名信號

支持自定義命名信號

支持與接收者之間的持久連接與短暫連接

通過弱引用實現與接收者之間的自動斷開連接

支持發送任意大小的數據

支持收集信號接收者的返回值

線程安全

創建信號

信號通過signal()方法進行創建：

>>> from blinker import signal
>>> initialized = signal("initialized")
>>> initialized is signal("initialized")
True

每次調用signal("name")都會返回同一個信號對象。因此這里signal()方法使用了單例模式。

訂閱信號

使用Signal.connect()方法注冊一個函數，每當觸發信號的時候，就會調用該函數。該函數以觸發信號的對象作為參數，這個函數其實就是信號訂閱者。

>>> def subscriber(sender):
...     print("Got a signal sent by %r" % sender)
...
>>> ready = signal("ready")
>>> ready.connect(subscriber)

觸發信號

使用Signal.send()方法通知信號訂閱者。

下面定義類Processor，在它的go()方法中觸發前面聲明的ready信號，send()方法以self為參數，也就是說Processor的實例是信號的發送者。

>>> class Processor:
...    def __init__(self, name):
...        self.name = name
...
...    def go(self):
...        ready = signal("ready")
...        ready.send(self)
...        print("Processing.")
...        complete = signal("complete")
...        complete.send(self)
...
...    def __repr__(self):
...        return "" % self.name
...
>>> processor_a = Processor("a")
>>> processor_a.go()
Got a signal sent by 
Processing.

注意到go()方法中的complete信號沒？并沒有訂閱者訂閱該信號，但是依然可以觸發該信號。如果沒有任何訂閱者的信號，結果是什么信號也不會發送，而且Blinker內部對這種情況進行了優化，以盡可能的減少內存開銷。

訂閱特定的發布者

默認情況下，任意發布者觸發信號，都會通知訂閱者。可以給Signal.connect()傳遞一個可選的參數，以便限制訂閱者只能訂閱特定發送者。

>>> def b_subscriber(sender):
...     print("Caught signal from processor_b.")
...     assert sender.name == "b"
...
>>> processor_b = Processor("b")
>>> ready.connect(b_subscriber, sender=processor_b)

現在訂閱者只訂閱了processor_b發布的ready信號:

>>> processor_a.go()
Got a signal sent by 
Processing.
>>> processor_b.go()
Got a signal sent by 
Caught signal from processor_b.
Processing.

通過信號收發數據

可以給send()方法傳遞額外的關鍵字參數，這些參數會傳遞給訂閱者。

>>> send_data = signal("send-data")
>>> @send_data.connect
... def receive_data(sender, **kw):
...     print("Caught signal from %r, data %r" % (sender, kw))
...     return "received!"
...
>>> result = send_data.send("anonymous", abc=123)
Caught signal from "anonymous", data {"abc": 123}

send()方法的返回值收集每個訂閱者的返回值，拼接成一個元組組成的列表。每個元組的組成為(receiver function, return value)。

匿名信號

前面我們創建的信號都是命名信號，每次調用Signal構造器都會創建一個唯一的信號,，也就是說每次創建的信號是不一樣的。下面對前面的Processor類進行改造，將signal作為它的類屬性。

>>> from blinker import Signal
>>> class AltProcessor:
...    on_ready = Signal()
...    on_complete = Signal()
...
...    def __init__(self, name):
...        self.name = name
...
...    def go(self):
...        self.on_ready.send(self)
...        print("Alternate processing.")
...        self.on_complete.send(self)
...
...    def __repr__(self):
...        return "" % self.name
...

上面創建的就是匿名信號。on_ready與on_complete是兩個不同的信號。

使用修飾器訂閱信號

除了使用connect()方法訂閱信號之外，使用@connect修飾器可以達到同樣的效果。

>>> apc = AltProcessor("c")
>>> @apc.on_complete.connect
... def completed(sender):
...     print "AltProcessor %s completed!" % sender.name
...
>>> apc.go()
Alternate processing.
AltProcessor c completed!

盡管這樣用起來很方便，但是這種形式不支持訂閱指定的發送者。這時，可以使用connect_via()：

>>> dice_roll = signal("dice_roll")
>>> @dice_roll.connect_via(1)
... @dice_roll.connect_via(3)
... @dice_roll.connect_via(5)
... def odd_subscriber(sender):
...     print("Observed dice roll %r." % sender)
...
>>> result = dice_roll.send(3)
Observed dice roll 3.

優化信號發送

信號通常會進行優化，以便快速的發送。不管有沒有訂閱者，都可以發送信號。如果發送信號時需要傳送的參數要計算很長時間，可以在發送之前使用receivers屬性先檢查一下是否有訂閱者。

>>> bool(signal("ready").receivers)
True
>>> bool(signal("complete").receivers)
False
>>> bool(AltProcessor.on_complete.receivers)
True

還可以檢查訂閱者是否訂閱了某個具體的信號發布者。

>>> signal("ready").has_receivers_for(processor_a)
True

GPU云服務器云服務器 blinker 深度學習庫機器學習庫 c機器學習庫

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/38026.html

ESP8266+Blinker 的萬物互聯（智能家居篇）

摘要：下面是演示視頻物聯網文章目錄前言一準備二操作步驟配置配置三效果展示總結前言實現萬物互聯第一步，千里點燈，然后添加你想要的模塊就可以了，接下來我們進入正文。前面學習...

CntChen 2021-11-24 09:38 評論0 收藏0
ESP8266-NodeMCU項目（三）：ESP8266-NodeMCU+Blinker+紅外模塊（

摘要：紅外模塊接入板子后，可進行以上操作，并接入，通過控制空調。材料清單板子紅外接收模塊紅外發射模塊線杜邦線左邊是紅外發射模塊，右邊是紅外接收模塊。錯開の折騰經歷：ES...

luoyibu 2021-10-08 10:04 評論0 收藏0
ESP8266-NodeMCU項目（四）：將上一項目的空調控制接入小愛同學（Blinker_APP同

摘要：錯開折騰經歷文章目錄前言思路設備配置程序代碼小愛訓練測試總結前言之前已經進行了項目二接入小愛同學以及項目三空調控制，接下來便是將二者合一，實現控制，溫濕度查看，以及小愛同學進行簡單空調控制。 ...

Loong_T 2021-10-09 09:43 評論0 收藏0
Flask Web Development —— 大型應用程序結構（下）

摘要：單元測試這個應用非常小以至于不需要太多的測試，但是作為示例會在示例中展示兩個簡單的測試定義。示例單元測試編寫好的測試使用的是來自于標準庫中標準的包。為了運行單元測試，可以在腳本中增加一個自定義的命令。 4、啟動腳本頂層目錄中的manage.py文件用于啟動應用。這個腳本會在示例7-8中展示。示例7-8. manage.py：啟動腳本 #!/usr/bin/env python im...

whidy 2019-07-31 11:35 評論0 收藏0
BLIINKER+ESP32編譯問題：some warnings being treated as

摘要：問題描述我在學習的時候想嘗試一下物聯網的使用，在調用官方的例程的時候發現程序編譯出現了問題上圖中的對應多個庫沒有問題，對實際編譯無影響。主要是出現了這個問題上圖中是的編譯器，出現問題是因為編譯器把一些警告當成錯誤了。再次編譯，編譯成功問題描述： ? ? ? ? 我在學習esp32的時候想嘗...

codecook 2021-10-13 09:39 評論0 收藏0