资讯专栏INFORMATION COLUMN

上云采购季！| 爆款云服务器低至4.6元/月，首单享0.9折

免费试用
首页/文章专栏/python blinker库学习

python blinker库学习

legendmohe / 1035人阅读

摘要：的信号机制就是基于它建立的。触发信号使用方法通知信号订阅者。每个元组的组成为。与是两个不同的信号。这时，可以使用优化信号发送信号通常会进行优化，以便快速的发送。

参考  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
服务器托管 专线服务 blinker 深度学习库 机器学习库 c机器学习库

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/38026.html

相关文章

  • ESP8266+Blinker 的万物互联（智能家居篇）

    摘要：下面是演示视频物联网文章目录前言一准备二操作步骤配置配置三效果展示总结前言实现万物互联第一步，千里点灯，然后添加你想要的模块就可以了，接下来我们进入正文。 前面学习...

    CntChen 评论0 收藏0

  • ESP8266-NodeMCU项目（三）：ESP8266-NodeMCU+Blinker+红外模块（

    摘要：红外模块接入板子后，可进行以上操作，并接入，通过控制空调。材料清单板子红外接收模块红外发射模块线杜邦线左边是红外发射模块，右边是红外接收模块。 错开の折腾经历：ES...

    luoyibu 评论0 收藏0

  • ESP8266-NodeMCU项目（四）：将上一项目的空调控制接入小爱同学（Blinker_APP同

    摘要：错开折腾经历文章目录前言思路设备配置程序代码小爱训练测试总结前言之前已经进行了项目二接入小爱同学以及项目三空调控制，接下来便是将二者合一，实现控制，温湿度查看，以及小爱同学进行简单空调控制。 ...

    Loong_T 评论0 收藏0

  • Flask Web Development —— 大型应用程序结构（下）

    摘要：单元测试这个应用非常小以至于不需要太多的测试，但是作为示例会在示例中展示两个简单的测试定义。示例单元测试编写好的测试使用的是来自于标准库中标准的包。为了运行单元测试，可以在脚本中增加一个自定义的命令。 4、启动脚本 顶层目录中的manage.py文件用于启动应用。这个脚本会在示例7-8中展示。 示例7-8. manage.py：启动脚本 #!/usr/bin/env python im...

    whidy 评论0 收藏0

  • BLIINKER+ESP32编译问题：some warnings being treated as

    摘要：问题描述我在学习的时候想尝试一下物联网的使用，在调用官方的例程的时候发现程序编译出现了问题上图中的对应多个库没有问题，对实际编译无影响。主要是出现了这个问题上图中是的编译器，出现问题是因为编译器把一些警告当成错误了。再次编译，编译成功 问题描述：         我在学习esp32的时候想尝...

    codecook 评论0 收藏0

发表评论

登陆后可评论

0条评论

legendmohe

男|高级讲师
我要私信

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<