【一】引入
- Flask框架中的信号基于blinker，其主要就是让开发者可是在flask请求过程中定制一些用户行为
【二】安装
【1】安装
pip3 install blinker
【2】内置信号
request_started = _signals.signal('request-started') # 请求到来前执行
request_finished = _signals.signal('request-finished') # 请求结束后执行
before_render_template = _signals.signal('before-render-template') # 模板渲染前执行
template_rendered = _signals.signal('template-rendered') # 模板渲染后执行
got_request_exception = _signals.signal('got-request-exception') # 请求执行出现异常时执行
request_tearing_down = _signals.signal('request-tearing-down') # 请求执行完毕后自动执行（无论成功与否）
appcontext_tearing_down = _signals.signal('appcontext-tearing-down')# 应用上下文执行完毕后自动执行（无论成功与否）
appcontext_pushed = _signals.signal('appcontext-pushed') # 应用上下文push时执行
appcontext_popped = _signals.signal('appcontext-popped') # 应用上下文pop时执行
message_flashed = _signals.signal('message-flashed') # 调用flask在其中添加数据时，自动触发
【3】使用信号
from flask import Flask,signals,render_template
app = Flask(__name__)
# 往信号中注册函数
def func(*args,**kwargs):
print('触发型号',args,kwargs)
signals.request_started.connect(func)
# 触发信号： signals.request_started.send()
@app.before_first_request
def before_first1(*args,**kwargs):
pass
@app.before_first_request
def before_first2(*args,**kwargs):
pass
@app.before_request
def before_first3(*args,**kwargs):
pass
@app.route('/',methods=['GET',"POST"])
def index():
print('视图')
return render_template('index.html')
if __name__ == '__main__':
app.wsgi_app
app.run()
【三】使用步骤
【1】引入
from flask import Flask, render_template
app = Flask(__name__, template_folder='templates')
@app.route('/', methods=['GET', 'POST'])
def index():
# 当我们想在模版渲染之前执行某段代码时，只能在这加，一个函数加一次，很浪费精力和时间
return render_template('index.html')
if __name__ == '__main__':
app.run()
- 使用内置信号优化上述问题
【2】使用内置信号
from flask import Flask, render_template, signals
app = Flask(__name__, template_folder='templates')
app.debug = True
@app.route('/', methods=['GET', 'POST'])
def index():
# 当我们想在模版渲染之前执行某段代码时，只能在这加，一个函数加一次，很浪费精力和时间
return render_template('index.html')
# (1) 使用内置信号，每次模板渲染之前都执行某段代码
# 第一步 ：写一个函数
def test(app, *args, **kwargs):
print('app :>>> ', app)
print('args :>>> ', args)
print('kwargs :>>> ', kwargs)
# 请求地址是根路径，才记录日志，其他不记录
if kwargs.get('context').get('request').path == '/':
print("记录日志 ... ")
print("模板渲染开始 --- ")
...
# (2) 绑定内置信号
signals.before_render_template.connect(test)
# （3）等待信号被触发--->只要执行到内置信号位置，绑定的函数就会执行
if __name__ == '__main__':
app.run()
- 启动项目
app :>>> <Flask '01'>
args :>>> ()
kwargs :>>> {'template': <Template 'index.html'>, 'context': {'g': <flask.g of '01'>, 'request': <Request 'http://127.0.0.1:5000/' [GET]>, 'session': <NullSession {}>}}
记录日志 ...
模板渲染开始 ---
【3】内置信号说明
request_started
request_started是一个信号，在每个请求到来之前执行。- 可以通过连接到这个信号来执行一些初始化操作或记录日志。
request_finished
request_finished是一个信号，在每个请求结束后执行。- 可以通过连接到这个信号来进行一些清理操作或处理请求完成后的逻辑。
before_render_template
before_render_template是一个信号，在模板渲染之前执行。- 可以通过连接到这个信号来修改要渲染的模板或添加一些数据。
template_rendered
template_rendered是一个信号，在模板渲染之后执行。- 可以通过连接到这个信号来进行一些后处理操作。
got_request_exception
got_request_exception是一个信号，在请求执行过程中出现异常时执行。- 可以通过连接到这个信号来处理请求异常并记录错误信息。
request_tearing_down
request_tearing_down是一个信号，会在请求执行完毕后自动执行，无论请求成功与否。- 可以通过连接到这个信号来进行一些收尾工作或资源释放操作。
appcontext_tearing_down
appcontext_tearing_down是一个信号，会在应用上下文执行完毕后自动执行，无论成功与否。- 可以通过连接到这个信号来进行一些应用上下文的收尾工作或资源释放操作。
appcontext_pushed
appcontext_pushed是一个信号，在应用上下文被push时执行。- 可以通过连接到这个信号来执行一些与应用上下文相关的操作。
appcontext_popped
appcontext_popped是一个信号，在应用上下文被pop时执行。- 可以通过连接到这个信号来执行一些与应用上下文相关的清理操作。
message_flashed
message_flashed是一个信号，在调用Flask中的消息闪现（flash）方法时自动触发。- 可以通过连接到这个信号来做一些闪现消息的处理逻辑。
【四】信号的自定义
【1】自定义信号
1 第一步：定义一个自定义 信号
2 第二步：写个函数
3 第三步：函数跟自己定义信号绑定
4 第四步：触发自定义信号---》我们做
- 自定义信号是在编程中常见的一种机制，它允许在特定情况下发送信号以通知其他部分执行相应的操作。
- 以下是一个详细解释和示例：
第一步：定义一个自定义信号
- 在Python编程语言中，我们可以使用QObject类的子类来定义自定义信号。
- 首先，我们需要导入PyQt库（如果未安装，请先使用pip进行安装）：
from PyQt5.QtCore import QObject, pyqtSignal
- 然后，创建一个新的类并继承自QObject类，这个类将作为信号的发出者：
class MyObject(QObject):
# 定义一个自定义信号
custom_signal = pyqtSignal(str)
- 在上述代码中，我们创建了一个名为"custom_signal"的自定义信号，它可以接受一个字符串参数。
第二步：写个函数
- 接下来，我们需要编写一个函数，该函数将在触发自定义信号时执行：
def do_something(self):
# 执行一些操作...
# 在适当的地方发送自定义信号，通过emit方法发送信号并传递参数
self.custom_signal.emit("自定义信号已被触发")
- 在上述代码中，我们定义了一个名为"do_something"的函数，并在适当的地方使用
self.custom_signal.emit()语句来触发自定义信号，并传递一个字符串参数。
第三步：函数跟自己定义信号绑定
- 为了确保函数和自定义信号相关联，我们需要在信号发出者的代码中进行连接。
- 通常，在类的初始化方法中进行这项操作：
class MyObject(QObject):
custom_signal = pyqtSignal(str)
def __init__(self):
super().__init__()
# 将自定义信号与函数关联
self.custom_signal.connect(self.handle_custom_signal)
- 在上述代码中，我们在初始化方法
__init__()中使用self.custom_signal.connnect()语句将自定义信号与处理函数handle_custom_signal()进行连接。
第四步：触发自定义信号
- 一旦自定义信号与函数进行了连接，我们就可以在需要的时候触发自定义信号了。
- 例如，在某个特定的条件下，我们可以调用"do_something()"函数来触发自定义信号：
my_object = MyObject()
my_object.do_something()
-
上述代码中，我们创建了一个"MyObject"实例，并通过调用"do_something()"方法来触发自定义信号。
-
当自定义信号被触发时，与之连接的处理函数将会执行。
-
在上述示例中，我们可以编写一个名为"handle_custom_signal()"的函数来处理自定义信号：
def handle_custom_signal(self, message):
print("接收到自定义信号:", message)
# 执行其他相应的操作...
- 上述代码中的"handle_custom_signal()"函数将被调用，并打印接收到的自定义信号的消息。
【2】flask自定义信号
1 第一步：定义一个自定义 信号
2 第二步：写个函数
3 第三步：函数跟自己定义信号绑定
4 第四步：触发自定义信号---》我们做
第一步：定义一个自定义信号
- 要在Flask中定义自定义信号，我们可以使用Python标准库中的
signals模块。 - 首先，导入相关的依赖项：
from blinker import Namespace
# 创建一个信号命名空间
my_signals = Namespace()
# 定义一个自定义信号
custom_signal = my_signals.signal('custom-signal')
- 在上述代码中，我们创建了一个信号命名空间
my_signals，然后使用my_signals.signal()方法定义了一个名为'custom-signal'的自定义信号。
第二步：写个函数
- 接下来，我们需要编写一个函数，该函数将在触发自定义信号时执行：
def handle_custom_signal(sender, **kwargs):
# 执行一些操作...
print("自定义信号被触发")
# 将函数注册为自定义信号的处理函数
custom_signal.connect(handle_custom_signal)
- 在上述代码中，我们定义了一个名为
handle_custom_signal()的函数来处理自定义信号。 - 通过使用
custom_signal.connect()方法，我们将该函数注册为自定义信号custom_signal的处理函数。
第三步：函数跟自己定义信号绑定
- 为了确保函数与自定义信号相关联，我们已经在第二步中使用
custom_signal.connect()方法进行了连接。
第四步：触发自定义信号
- 一旦自定义信号与函数进行了连接，我们就可以在需要的时候触发自定义信号了。
- 例如，在Flask应用程序的某个特定操作中，我们可以使用以下代码触发自定义信号：
custom_signal.send(current_app._get_current_object())
- 在上述代码中，
custom_signal.send()方法用于触发自定义信号custom_signal。- 我们通过
current_app._get_current_object()获取当前Flask应用程序的实例，并将其作为发送者。
- 我们通过
- 当自定义信号被触发时，与之连接的处理函数将会执行。
- 在上述示例中，我们定义的
handle_custom_signal()函数将被调用，并打印"自定义信号被触发"的消息。
- 在上述示例中，我们定义的
【六】flask自定义信号演示
from flask import Flask, session, render_template, signals
from flask.signals import _signals
# pip3 install blinker
app = Flask(__name__)
app.debug = True
app.secret_key = 'asdfasdfasdf'
#### 自定义信号---》session每次放一个值，我们就执行信号
# 1 定义信号
# 自定义信号
session_input = _signals.signal('session_input')
# 2 写个函数
def test2(*args, **kwargs):
print(args) # app
print(kwargs) # {session kk}
print('session放值了')
if kwargs.get('kk').get('name') == 'xx':
print('记录日志')
# 3 绑定信号
session_input.connect(test2)
# 4 触发信号
@app.route('/')
def index():
session['uu'] = '00'
session_input.send(app, session=session, kk={'name': 'uu'})
return render_template('index.html')
@app.route('/home')
def home():
return render_template('home.html')
@app.route('/order')
def order():
session['xx'] = 'xx'
session_input.send(app, session=session, kk={'name': 'xx'})
return "order"
if __name__ == '__main__':
app.run(port=8080)
【七】信号的使用
- 记录日志：只要是张三用户，访问index页面，就记录日志
- 只要用户表中，插入一条记录，就给用户发个短信通知
- User.object.create--->调用发短信方法--》找到10个地址--》改10个地方
- 如果有个内置信号---》只要表中增加记录，就会触发这个信号----》通过信号内判断这个表是不是User表，决定要不要发短信
- flask中没有这个内置信号---》自定义
信号的使用在软件开发中是一种重要的机制，它可以帮助我们对系统中发生的特定事件作出响应。根据您提供的内容，我将进一步扩充解释信号的使用，并给出一些案例来说明。
【1】记录日志
- 首先，记录日志是一个常见的需求，通过使用信号，我们可以在用户访问index页面时自动记录日志。
- 下面是一个使用Django框架的例子：
from django.dispatch import receiver
from django.contrib.auth.models import User
from django.dispatch import Signal
# 创建一个信号
user_visited_index = Signal(providing_args=['user'])
# 定义信号的处理函数
@receiver(user_visited_index)
def log_user_visit(sender, **kwargs):
user = kwargs['user']
# 在这里进行日志记录操作，比如保存到数据库或写入日志文件
# 触发信号
user = User.objects.get(username='张三')
user_visited_index.send(sender=None, user=user)
- 以上代码中，我们首先创建了一个名为
user_visited_index的信号，定义了一个user参数用于传递用户对象。 - 然后，我们通过
@receiver装饰器将log_user_visit函数与信号进行绑定，该函数负责处理信号触发时的逻辑，比如记录用户访问日志。 - 最后，通过
send()方法触发信号，并传递相应的参数。
【2】用户发送短信通知
- 接下来，让我们来看一个给用户发送短信通知的案例。
- 当在用户表中插入一条记录时，通过信号触发发送短信通知的操作。
- 以下是一个使用Django框架的例子：
from django.dispatch import receiver
from django.db.models.signals import post_save
from django.contrib.auth.models import User
# 定义信号的处理函数
@receiver(post_save, sender=User)
def send_sms_notification(sender, instance, created, **kwargs):
if created:
# 只在创建新用户时发送短信通知
user = instance
# 在这里调用发短信的方法，给用户发送短信通知
- 在上述代码中，我们使用
post_save信号来监听User模型的保存操作。 - 当有新的用户记录被创建时，
send_sms_notification函数将被自动调用，我们可以在其中编写发送短信的代码逻辑。
【七】Django中的信号
- Django中的信号是一种机制，用于在特定事件发生时自动触发相关的操作或函数。
- 通过使用信号，可以实现模块间的解耦和事件驱动的编程。
- 在Django中，有两种类型的信号：内置信号和自定义信号。
【1】内置信号
- Django提供了许多内置信号，以便我们在与数据库交互、处理请求和响应等方面执行特定的操作。
#Model signals
pre_init # django的modal执行其构造方法前，自动触发
post_init # django的modal执行其构造方法后，自动触发
pre_save # django的modal对象保存前，自动触发
post_save # django的modal对象保存后，自动触发
pre_delete # django的modal对象删除前，自动触发
post_delete # django的modal对象删除后，自动触发
m2m_changed # django的modal中使用m2m字段操作第三张表（add,remove,clear）前后，自动触发
class_prepared # 程序启动时，检测已注册的app中modal类，对于每一个类，自动触发
Management signals
pre_migrate # 执行migrate命令前，自动触发
post_migrate # 执行migrate命令后，自动触发
Request/response signals
request_started # 请求到来前，自动触发
request_finished # 请求结束后，自动触发
got_request_exception # 请求异常后，自动触发
Test signals
setting_changed # 使用test测试修改配置文件时，自动触发
template_rendered # 使用test测试渲染模板时，自动触发
Database Wrappers
connection_created # 创建数据库连接时，自动触发
from django.db.models.signals import pre_save
from django.dispatch import receiver
@receiver(pre_save)
def my_callback(sender, **kwargs):
print("对象创建成功")
print(sender)
print(kwargs)
1. Model signals（模型信号）
-
pre_init：在执行模型的构造函数之前触发。这个信号可以在实例化一个模型对象之前执行一些操作，比如设置默认值。@receiver(pre_init, sender=MyModel) def pre_init_callback(sender, **kwargs): print("Before initializing MyModel instance.") -
post_init：在执行模型的构造函数之后触发。这个信号允许在实例化一个模型对象之后执行一些操作，比如初始化关联对象。@receiver(post_init, sender=MyModel) def post_init_callback(sender, instance, **kwargs): print(f"After initializing {instance} instance.") -
pre_save：在保存模型对象之前触发。这个信号可以用于在保存模型对象之前进行一些预处理或验证操作。@receiver(pre_save, sender=MyModel) def pre_save_callback(sender, instance, **kwargs): print(f"Before saving {instance} instance.") -
post_save：在保存模型对象之后触发。这个信号可以用于在保存模型对象之后执行一些后处理操作。@receiver(post_save, sender=MyModel) def post_save_callback(sender, instance, created, **kwargs): if created: print(f"New {instance} instance has been saved.") else: print(f"{instance} instance has been updated.") -
pre_delete：在删除模型对象之前触发。这个信号可以用于执行一些与删除关联的操作。@receiver(pre_delete, sender=MyModel) def pre_delete_callback(sender, instance, **kwargs): print(f"Before deleting {instance} instance.") -
post_delete：在删除模型对象之后触发。这个信号可以用于执行一些与删除关联的操作。@receiver(post_delete, sender=MyModel) def post_delete_callback(sender, instance, **kwargs): print(f"{instance} instance has been deleted.") -
m2m_changed：当使用Many-to-Many字段操作第三张中间表时触发，比如add、remove和clear等操作。这个信号可以用于在修改Many-to-Many关系时进行额外的处理操作。@receiver(m2m_changed, sender=MyModel.m2m_field.through) def m2m_changed_callback(sender, instance, action, reverse, **kwargs): if action == "pre_add": print(f"Adding related objects to {instance}.") elif action == "pre_remove": print(f"Removing related objects from {instance}.") # 其他操作类似 -
class_prepared：在程序启动时检测已注册的应用程序中的模型类，并为每个类触发一次。这个信号可以用于在模型类准备就绪后执行一些初始化操作。@receiver(class_prepared, sender=MyModel) def class_prepared_callback(sender, **kwargs): print(f"{sender} class is prepared and ready to use.")
2. Management signals（管理信号）
-
pre_migrate：在执行migrate命令之前触发。这个信号可以用于在执行数据库迁移之前执行一些自定义操作。@receiver(pre_migrate) def pre_migrate_callback(sender, app_config, **kwargs): print(f"Preparing to migrate {app_config.label}.") -
post_migrate：在执行migrate命令之后触发。这个信号可以用于在执行数据库迁移之后执行一些自定义操作。@receiver(post_migrate) def post_migrate_callback(sender, app_config, **kwargs): print(f"{app_config.label} has been migrated successfully.")
3. Request/response signals（请求/响应信号）
-
request_started：在请求到达之前触发。这个信号可以用于在处理请求之前执行一些操作，比如记录请求的开始时间。@receiver(request_started) def request_started_callback(sender, environ, **kwargs): print("Request started.") -
request_finished：在请求处理完毕后触发。这个信号可以用于在每个请求处理完毕后执行一些操作，比如记录请求的结束时间。@receiver(request_finished) def request_finished_callback(sender, **kwargs): print("Request finished.") -
got_request_exception：在请求处理过程中出现异常时触发。这个信号可以用于在处理请求异常时执行一些额外的操作。@receiver(got_request_exception) def got_request_exception_callback(sender, request, **kwargs): print(f"Exception occurred while processing request: {kwargs.get('exception')}.")
4. Test signals（测试信号）
-
setting_changed：在使用测试修改配置文件时触发。这个信号可以用于在进行测试期间管理配置更改，并对更改进行特定的响应。@receiver(setting_changed) def setting_changed_callback(sender, setting, value, enter, **kwargs): if enter: print(f"Setting '{setting}' changed to '{value}'.") else: print(f"Setting '{setting}' restored to its original value.") -
template_rendered：在使用测试渲染模板时触发。这个信号可以用于在进行模板渲染测试时执行一些验证或记录操作。@receiver(template_rendered) def template_rendered_callback(sender, template, content, **kwargs): print(f"Template '{template.name}' has been rendered with content: {content}.")
5. Database Wrappers（数据库包装器）
connection_created：在创建数据库连接时触发。这个信号可以用于在每次创建数据库连接时执行一些操作，比如设置连接特定的选项。@receiver(connection_created) def connection_created_callback(sender, connection, **kwargs): print("Database connection created.")
通过使用@receiver装饰器并定义相应的回调函数，我们可以将这些信号与特定的操作关联起来，以便在信号触发时执行相应的逻辑。以上就是对不同类型信号的扩展和解释，以及相关示例。
【2】内置信号使用(当user表创建用户，就给用户发个邮件)
1 写个函数 #放到__init__里
from django.db.models.signals import pre_save
import logging
def callBack(sender, **kwargs):
logging.debug('%s创建了一个%s对象'%(sender._meta.model_name,kwargs.get('instance').title))
2 绑定内置信号
pre_save.connect(callBack)
3 等待触发
# 在__init__.py文件中添加以下代码
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User
from django.core.mail import send_mail
@receiver(post_save, sender=User)
def send_email_on_user_creation(sender, instance, created, **kwargs):
if created:
# 发送电子邮件给新创建的用户
subject = '欢迎加入我们的网站'
message = '您已成功注册我们的网站。感谢您的加入！'
from_email = '[email protected]'
to_email = instance.email
send_mail(subject, message, from_email, [to_email])
- 在上述代码中，我们定义了一个名为
send_email_on_user_creation的回调函数，它会在用户模型（User）保存后触发。- 在回调函数中，我们通过
send_mail函数向新创建的用户发送欢迎邮件。
【3】自定义信号
-
除了内置信号，Django还支持自定义信号。
- 通过自定义信号，我们可以在应用程序中定义自己的事件，并根据需要触发和处理这些事件。
-
下面是自定义信号的基本步骤：
-
定义信号：在某个文件中定义信号对象，并指定传递的参数（可以是任意数量的参数）。
-
注册信号：编写接收信号的回调函数，并使用
connect方法将其连接到信号上。 -
触发信号：在适当的时候，使用
send方法触发信号并传递相应的参数。
-
# myapp/signals.py 文件
from django.dispatch import Signal
pizza_done = Signal(providing_args=['toppings', 'size'])
# views.py 文件
from myapp.signals import pizza_done
def order_pizza(request):
# ... 执行订购披萨的逻辑
toppings = ['pepperoni', 'mushrooms']
size = 'large'
# 披萨制作完成后触发信号
pizza_done.send(sender='myapp', toppings=toppings, size=size)
# signals.py 文件
from myapp.signals import pizza_done
def callback(sender, **kwargs):
print('Pizza is ready!')
print('Toppings:', kwargs['toppings'])
print('Size:', kwargs['size'])
# 将回调函数连接到自定义信号
pizza_done.connect(callback)
- 在上述示例中，我们首先定义了一个名为
pizza_done的自定义信号对象，并指定它可以传递两个参数：toppings和size。 - 然后，在
order_pizza函数中，当披萨制作完成后，我们使用pizza_done.send方法触发信号，并传递相关的参数。 - 最后，我们编写了一个名为
callback的回调函数，并将其连接到自定义信号pizza_done上。
#1 定义信号（一般创建一个py文件）（toppings，size 是接受的参数）
import django.dispatch
pizza_done = django.dispatch.Signal(providing_args=["toppings", "size"])
# 2 注册信号
def callback(sender, **kwargs):
print("callback")
print(sender,kwargs)
pizza_done.connect(callback)
# 3 触发信号
from 路径 import pizza_done
pizza_done.send(sender='seven',toppings=123, size=456)
【4】自定义信号的应用场景
（1）缓存更新与双写一致性
- 在分布式系统中，缓存是提高性能和减少对后端资源压力的常见手段。
- 当数据更新时，需要及时更新缓存以保持数据的一致性。
- 使用自定义信号可以实现双写一致性，确保数据在更新完成前，缓存已经被刷新。
- 例如，在电子商务平台中，当用户下单购买商品时，除了更新数据库中的订单信息，还可以发送一个自定义信号，通知缓存系统更新该订单相关信息，以避免缓存数据与数据库数据不一致。
（2）异步任务处理
- 在复杂的系统中，有许多需要异步执行的任务，如发送邮件、生成报表、处理图像等。
- 通过自定义信号，可以触发异步任务并将结果通知给相应的模块。
- 例如，在一个社交媒体平台上发布帖子时，可以异步处理图片的上传和压缩，并通过自定义信号将处理结果返回给前端页面，使用户能够即时查看上传图片的处理状态。
（3）事件驱动的系统架构
- 自定义信号还可以用于实现事件驱动的系统架构。
- 当某个事件发生时，可以发送相应的自定义信号，触发事件处理器执行相应的逻辑。
- 例如，在一个在线支付系统中，当用户完成支付操作时，可以发送一个自定义信号，通知相应的模块进行订单处理、库存更新等操作。
（4）状态同步与通知
- 自定义信号还可以用于状态同步和通知。
- 当多个模块之间需要共享某个状态，并保持实时更新时，可以使用自定义信号来实现状态的同步和通知。
- 例如，在一个即时通讯软件中，当用户上线或下线时，可以通过发送自定义信号来通知好友列表进行状态的更新，以便及时显示对方的在线状态。
【5】自定义信号的应用场景分析
1. 缓存更新与双写一致性
- 在分布式系统中，为了提高性能和减少对后端资源的压力，常常使用缓存来保存热门的数据。
- 然而，当数据发生更新时，需要确保缓存与数据库的数据保持一致，避免脏读或不一致的情况发生。
- 自定义信号可以用于实现双写一致性，确保缓存在更新完成前已经被刷新。
- 假设有一个电子商务平台，用户通过下单购买商品，需要更新订单信息同时更新缓存。
- 具体步骤如下：
场景需求分析：
- 用户下单购买商品时，首先会将订单信息写入数据库。
- 同时，发送一个自定义信号，通知缓存系统更新该订单相关信息。
代码演示：
# 示例代码中使用了Python的signal模块，来实现自定义信号
import signal
def update_cache(signal, frame):
# 更新缓存的逻辑，将数据库中对应订单的缓存数据进行刷新
pass
def handle_order():
# 处理用户下单的逻辑，包括写入数据库等操作
# ...
# 发送自定义信号，触发更新缓存的操作
signal.alarm(1)
# 注册自定义信号的处理函数
signal.signal(signal.SIGALRM, update_cache)
# 用户下单购买商品时调用handle_order进行处理
handle_order()
2. 异步任务处理
- 复杂的系统中存在许多需要异步执行的任务，如邮件发送、报表生成、图像处理等。
- 通过自定义信号，可以触发异步任务，并将结果通知给相应的模块。
- 假设在一个社交媒体平台上发布帖子时，需要异步处理上传的图片，包括图片的上传和压缩，并通过自定义信号将处理结果返回给前端页面，以使用户能即时查看图片处理状态。
场景需求分析：
- 用户在发布帖子时上传图片，需要对图片进行异步处理。
- 处理完成后，用自定义信号通知前端页面并显示图片的处理状态。
代码演示：
import signal
def handle_image_processing(signal, frame):
# 图片处理的具体逻辑，例如上传、压缩等操作
pass
def handle_post():
# 帖子处理的逻辑，包括获取用户提交的数据等操作
# ...
# 发送自定义信号，触发图片处理操作
signal.alarm(1)
# 注册自定义信号的处理函数
signal.signal(signal.SIGALRM, handle_image_processing)
# 用户发布帖子时调用handle_post进行处理
handle_post()
3. 事件驱动的系统架构
- 自定义信号还可以用于实现基于事件驱动的系统架构。
- 当某个事件发生时，通过发送相应的自定义信号，触发事件处理器执行相应的逻辑。
- 假设一个在线支付系统中，用户完成支付操作时，需要通知相应的模块进行订单处理和库存更新等操作。
场景需求分析：
- 在用户完成支付操作时，通过发送自定义信号通知相关模块进行订单处理和库存更新。
代码演示：
import signal
def handle_payment(signal, frame):
# 进行订单处理、库存更新等操作
pass
def handle_checkout():
# 处理用户支付的逻辑，包括获取支付信息等操作
# ...
# 发送自定义信号，触发相关模块进行后续处理
signal.alarm(1)
# 注册自定义信号的处理函数
signal.signal(signal.SIGALRM, handle_payment)
# 用户支付操作完成时调用handle_checkout进行处理
handle_checkout()
4. 状态同步与通知
- 自定义信号还可以用于状态同步和通知的场景。
- 当多个模块之间需要共享某个状态，并保持实时更新时，可以使用自定义信号来实现状态的同步和通知。
- 假设在一个即时通讯软件中，当用户上线或下线时，通过发送自定义信号来通知好友列表进行状态更新，以便及时显示对方的在线状态。
场景需求分析：
- 在用户上线或下线时，发送自定义信号通知好友列表进行状态的更新。
代码演示：
import signal
def update_status(signal, frame):
# 更新状态的逻辑，通常会从消息队列等数据源获取实时状态信息并进行更新
pass
def handle_user_status_change():
# 处理用户上线/下线的逻辑，包括更新用户状态等操作
# ...
# 发送自定义信号，触发好友列表的状态更新
signal.alarm(1)
# 注册自定义信号的处理函数
signal.signal(signal.SIGALRM, update_status)
# 用户上线/下线时调用handle_user_status_change进行处理
handle_user_status_change()
【补充】信号量
【一】理解方式一
- 同进程的一样，Semaphore管理一个内置的计数器
- 每当调用acquire()时内置计数器-1；
- 调用release() 时内置计数器+1；
- 计数器不能小于0；
- 当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。
- 实例：
- (同时只有5个线程可以获得semaphore,即可以限制最大连接数为5)：
from threading import Thread,Semaphore
import threading
import time
# def func():
# if sm.acquire():
# print (threading.currentThread().getName() + ' get semaphore')
# time.sleep(2)
# sm.release()
def func():
sm.acquire()
print('%s get sm' %threading.current_thread().getName())
time.sleep(3)
sm.release()
if __name__ == '__main__':
sm=Semaphore(5)
for i in range(23):
t=Thread(target=func)
t.start()
- 与进程池是完全不同的概念，进程池Pool(4)，最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一堆线程/进程
【二】理解方式二
- 在Python中，信号量（Semaphore）是一种同步原语，用于控制对共享资源的访问。
- 它通常用于限制同时访问某个资源的线程或进程的数量。
- 信号量维护了一个内部计数器，可以通过修改计数器的值来控制访问资源的线程或进程数量。
- 当信号量的计数器大于0时，允许线程或进程访问资源；
- 当计数器等于0时，线程或进程将被阻塞，直到计数器的值大于0。
- 在Python中，信号量是通过
threading.Semaphore类实现的。下面是一个使用信号量的示例：
import threading
# 创建一个信号量，初始计数为3
semaphore = threading.Semaphore(3)
# 定义一个函数，模拟访问共享资源的操作
def access_resource():
# 获取信号量，如果计数器为0，则阻塞线程
semaphore.acquire()
try:
# 执行访问共享资源的操作
print("Accessing resource...")
# ...
finally:
# 释放信号量，增加计数器的值
semaphore.release()
# 创建多个线程进行资源访问
for i in range(5):
thread = threading.Thread(target=access_resource)
thread.start()
-
在上面的示例中，我们创建了一个初始计数为3的信号量
semaphore。- 然后，我们定义了一个名为
access_resource的函数，该函数模拟访问共享资源的操作。 - 在函数中，我们首先调用
semaphore.acquire()获取信号量，如果信号量的计数器为0，则阻塞当前线程。 - 然后，执行访问共享资源的操作。
- 最后，在
finally块中调用semaphore.release()释放信号量，增加计数器的值，让其他线程可以继续访问资源。
- 然后，我们定义了一个名为
-
在上述示例中，虽然有5个线程同时尝试访问资源，但由于信号量的计数器初始值为3，其中只有3个线程能够成功获取信号量，访问共享资源。
- 另外两个线程将被阻塞，直到有一个线程释放信号量。
-
通过使用信号量，我们可以灵活控制对共享资源的访问，并防止出现资源竞争和临界区问题。