Python 强大的信号库 blinker 入门详细教程

1. 什么是 blinker

blinker 是 Python 中一个强大的信号（Signal）处理库，它可以帮助我们更加方便地处理信号和槽机制，实现不同函数间数据传递，类似于事件驱动机制。

2. 安装 blinker

我们可以使用 pip 工具来安装 blinker 库：

pip install blinker

3. blinker 的基本使用

blinker 中最核心的概念是信号（Signal）和槽（Slot），信号代表一个事件，而槽则是一个函数，blinker 将信号与槽联系起来，信号发生时，对应的槽函数就会被调用。

我们先来看一个最简单的例子：

from blinker import signal

# 定义信号
hello_signal = signal('hello')

# 定义槽函数
def say_hello():
    print('Hello World!')

# 将信号与槽函数关联
hello_signal.connect(say_hello)

# 发送信号，即调用信号函数
hello_signal.send()

以上代码我们定义了一个名为 hello_signal 的信号，以及一个名为 say_hello 的槽函数，然后将两者关联起来。在最后一行代码中，我们发送了一个信号，触发了相应的槽函数。

输出结果如下：

Hello World!

可以看到，当我们发送信号后，say_hello 函数会被调用，并输出 Hello World! 字符串。

在 blinker 中，还可以定义带参数的信号，我们可以通过信号函数和槽函数的参数传递来传递参数。例如：

from blinker import signal

# 定义带参数的信号
hello_signal = signal('hello')

# 定义带参数的槽函数
def say_hello(name):
    print('Hello, %s!' % name)

# 将信号与槽函数关联
hello_signal.connect(say_hello)

# 发送信号，并传递参数
hello_signal.send(name='Tom')

在上述代码中，我们为信号和槽函数都定义了参数，然后在发送信号时，使用 send 方法传递了一个名为 name 的参数，输出结果为：

Hello, Tom!

4. 示例说明

示例一

我们假设我们正在开发一个 Python HTTP 服务，服务中需要处理用户提交的注册请求，并在注册成功后向用户发送一封电子邮件。在这种情况下，我们可以使用 blinker 来实现注册请求与发送电子邮件的两个操作之间的解耦。

在下面的示例代码中，我们将定义一个 register_signal 信号和一个 send_email 槽函数，并使用 connect 方法将二者关联起来。

import smtplib
from email.mime.text import MIMEText
from blinker import Namespace

my_signals = Namespace()

# 定义一个名为 register 的信号
register_signal = my_signals.signal('register')


# 定义一个发送电子邮件的槽函数
def send_email(sender, email_address):
    # 构建邮件内容
    msg = MIMEText('Thanks for register!')

    # 设置邮件的基本信息
    msg['Subject'] = 'Thanks for register'
    msg['From'] = '[e-mail]'
    msg['To'] = email_address

    # 创建 SMTP 连接
    s = smtplib.SMTP('smtp.163.com')

    # 登陆邮箱发送邮件
    s.login('[e-mail]', '************')
    s.sendmail('[e-mail]', [email_address], msg.as_string())
    s.quit()


# 将 register 信号与 send_email 槽函数关联起来
register_signal.connect(send_email, sender='blinker')


# 在提交注册请求时，触发 register 信号
def register(email):
    print('Get a register request from %s' % email)
    register_signal.send(email_address=email)


# 调用注册函数
register('[e-mail]')

在上述代码中，我们定义了一个名为 register 的函数，用于处理用户提交的注册请求。当有用户提交注册请求时，我们会调用 register 函数，此时会触发 register_signal 信号，并将注册用户的电子邮件地址作为参数传递给 send_email 槽函数。

send_email 函数会使用 SMTP 协议向注册用户发送一封电子邮件，邮件内容为“Thanks for register!”。

在最后一行代码中，我们调用了 register 函数并传入了一个测试用例电子邮件地址 [e-mail]，执行结果为：

Get a register request from [e-mail]

可以看到，我们成功触发了 register_signal 信号，并将 [e-mail] 地址传递给了 send_email 槽函数，同时程序还向该地址发送了一封电子邮件，达到了我们的预期效果。

示例二

在这个示例中，我们将定义一个名为 upload_signal 的信号，并使用 connect 方法将其关联到两个槽函数：print_file_name 和 backup_file。当 upload_signal 信号被触发时，这两个函数都会被调用。

from blinker import signal

# 定义 upload_signal 信号
upload_signal = signal('upload')


# 定义打印文件名的槽函数
def print_file_name(sender, file_name):
    print('New file uploaded: %s' % file_name)


# 定义备份文件的槽函数
def backup_file(sender, file_name):
    print('Back up file: %s' % file_name)


# 将 upload_signal 信号与两个槽函数关联起来
upload_signal.connect(print_file_name)
upload_signal.connect(backup_file)


# 上传文件时触发 upload_signal 信号
def upload_file(file_name):
    print('Uploading file: %s' % file_name)
    upload_signal.send(file_name=file_name)


# 调用上传文件函数
upload_file('hello.txt')

在上述示例代码中，我们定义了一个名为 upload_signal 的上传信号。当文件被上传时，我们会调用 upload_file 函数来触发该信号，并将上传文件的文件名作为参数传递给 upload_signal.send 方法。同时，我们还将该信号与两个槽函数 print_file_name 和 backup_file 关联起来，当信号被触发时，这两个函数都会被调用。

在最后一行代码中，我们调用 upload_file 函数并传入一个测试文件名 hello.txt，执行结果为：

Uploading file: hello.txt
New file uploaded: hello.txt
Back up file: hello.txt

可以看到，当我们上传了 hello.txt 文件后，upload_signal 信号被触发，同时 print_file_name 和 backup_file 函数都会被调用，分别输出了“New file uploaded: hello.txt”和“Back up file: hello.txt”这两个字符串，达到了我们的预期效果。