Flask 的信号（Signals）是什么？和 before_request 这类钩子有什么区别？

简化版

Flask 的信号基于 blinker 库，是一套「发布-订阅」机制：框架在关键时刻发出信号（request_started、request_finished、template_rendered、got_request_exception、appcontext_tearing_down 等），任何代码都可以订阅它们做出反应，而发送方完全不知道订阅方的存在。和 before_request 这类钩子最本质的区别是「能不能影响流程」：钩子是请求处理链条的一部分——before_request 返回非 None 就会中断请求直接返回响应，after_request 必须返回 response 且可以修改它；而信号是纯粹的「通知」，接收者的返回值会被完全忽略，无法中断流程也无法改写响应。所以定位很清晰：要控制流程用钩子，要旁路观察用信号（日志、埋点、监控、审计、缓存失效、发通知）。Flask 3.0 起 blinker 成为必需依赖（之前是可选的，没装时信号静默失效——这曾是个隐蔽的坑）。使用时有三个要点：① 用 signal.connect(fn, sender=app) 限定发送者，否则多 app 场景会串；② 保持一个对接收函数的强引用——blinker 默认用弱引用，局部函数被垃圾回收后订阅会静默消失，所以要么用模块级函数、要么传 weak=False；③ 接收函数里抛异常会影响发送方（send() 是同步调用的），所以要自己 try/except 兜住。核心记忆：信号 = 通知，改不了流程；钩子 = 流程的一环，能中断和改写；记得限定 sender 和防止弱引用被回收。

详细版

Flask 内置信号一览：

信号	触发时机	典型用途
`request_started`	请求上下文推入后、视图前	计时起点、埋点
`request_finished`	响应生成后	记录响应状态
`got_request_exception`	视图抛出未处理异常时	上报 Sentry
`request_tearing_down`	请求上下文销毁前	清理
`template_rendered`	模板渲染完成	测试断言
`before_render_template`	模板渲染前	注入变量
`appcontext_pushed/popped`	应用上下文进出	测试打桩
`appcontext_tearing_down`	应用上下文销毁	释放资源
`message_flashed`	调用 `flash()`	记录提示

# ① ★订阅内置信号★
from flask import request_started, request_finished, got_request_exception

def on_started(sender, **extra):          # ★sender 是 app 对象★
    g._t0 = time.perf_counter()

def on_finished(sender, response, **extra):   # ★★额外参数按信号而定★★
    dur = time.perf_counter() - g._t0
    current_app.logger.info("%s %s %s %.1fms",
                            request.method, request.path,
                            response.status_code, dur * 1000)

def on_exception(sender, exception, **extra):
    sentry_sdk.capture_exception(exception)

def init_signals(app):
    request_started.connect(on_started, app)      # ★★限定 sender=app★★
    request_finished.connect(on_finished, app)
    got_request_exception.connect(on_exception, app)

# ② ★装饰器写法★
@request_finished.connect_via(app)         # ★connect_via = 限定 sender★
def log_response(sender, response, **extra): ...

# ③ ★★自定义信号★★
from blinker import Namespace
_signals = Namespace()                     # ★用命名空间避免全局冲突★
order_paid = _signals.signal("order-paid")
user_registered = _signals.signal("user-registered")

# 发送方（业务代码，★不知道谁在监听★）
def pay_order(order):
    order.status = "paid"
    db.session.commit()
    order_paid.send(current_app._get_current_object(),   # ★★不能传 LocalProxy★★
                    order=order, amount=order.amount)

# 接收方（各自独立，★可以有多个★）
@order_paid.connect
def send_receipt(sender, order, **extra):
    mail.send(...)

@order_paid.connect
def update_stats(sender, order, amount, **extra):
    Stats.increment(amount)

@order_paid.connect
def clear_cache(sender, order, **extra):
    cache.delete_memoized(get_user_orders, order.user_id)

# ④ ★★弱引用陷阱★★
def setup(app):
    def handler(sender, **extra):          # ★局部函数★
        ...
    order_paid.connect(handler)            # ✗ ★函数出作用域后被 GC，订阅失效★
    order_paid.connect(handler, weak=False)  # ✓ ★保持强引用★
# ✓ 更好：★用模块级函数★

# ⑤ ★临时订阅（测试常用）★
with captured_signal(order_paid) as recorded:
    pay_order(order)
    assert len(recorded) == 1

from flask import template_rendered
@contextmanager
def captured_templates(app):
    recorded = []
    def record(sender, template, context, **extra):
        recorded.append((template, context))
    template_rendered.connect(record, app)
    try:
        yield recorded
    finally:
        template_rendered.disconnect(record, app)   # ★★记得断开★★

# ⑥ ★钩子 vs 信号的对照★
@app.before_request
def check_auth():
    if not logged_in():
        return redirect("/login")       # ★★返回非 None 会中断请求★★

@request_started.connect_via(app)
def on_start(sender, **extra):
    return redirect("/login")           # ★★返回值被忽略，什么也不会发生★★

⚠️ 三个必须记住的点：① 信号无法影响请求流程。before_request 返回非 None 值会让 Flask 跳过视图直接返回该响应（用于鉴权拦截、维护模式）；after_request 接收 response 并必须返回它（可以修改）。而信号接收者的返回值被 send() 收集但 Flask 完全忽略——你不能用信号做权限拦截，也不能用它改响应头。定位是「旁路观察」：日志、监控、埋点、审计、缓存失效、发通知。② blinker 默认使用弱引用——signal.connect(fn) 不会阻止 fn 被垃圾回收。如果 fn 是定义在函数内部的局部函数或 lambda，函数返回后它就被回收了，订阅静默消失且没有任何报错。解决办法：用模块级函数（最推荐）、或者传 weak=False、或者自己保存一份引用。这个坑的现象是「本地测试好好的，跑久了信号就不触发了」。③ send() 是同步的、在同一线程里顺序调用所有接收者——所以接收者里的耗时操作会直接拖慢请求，接收者里抛出的异常也会向上传播影响发送方（blinker 不吞异常）。要发邮件、调外部 API 就在接收者里投递到 Celery，同时每个接收者内部都要 try/except 兜住异常，避免「统计功能挂了导致下单失败」。

完整版教学

一、信号与钩子的本质区别

★ 请求处理链条上的位置：
  ┌──────────────────────────────────────────────────────────┐
  │ 请求进来                                                    │
  │   ↓ ★request_started 信号★（通知，无法干预）                 │
  │   ↓ ★before_request 钩子★（★返回非 None → 中断，直接响应★）   │
  │   ↓ url_map 匹配 → 视图函数                                 │
  │   ↓ （异常）★got_request_exception 信号★ → errorhandler      │
  │   ↓ ★after_request 钩子★（★接收并可修改 response★）           │
  │   ↓ ★request_finished 信号★（通知）                          │
  │   ↓ ★teardown_request 钩子★（★异常时也会执行★）               │
  │   ↓ ★request_tearing_down 信号★                             │
  │ 响应返回                                                    │
  └──────────────────────────────────────────────────────────┘

★ ★核心区别表★：
  ┌──────────────┬────────────────────┬──────────────────────┐
  │              │ ★钩子（hook）★      │ ★信号（signal）★      │
  ├──────────────┼────────────────────┼──────────────────────┤
  │ 本质          │ ★流程的一环★        │ ★旁路的通知★          │
  │ 返回值        │ ★有意义（能中断/改写）│ ★被忽略★             │
  │ 能否中断请求  │ ★✓ before_request★  │ ★✗★                  │
  │ 能否改响应    │ ★✓ after_request★   │ ★✗★                  │
  │ 数量          │ 有限的几个固定点     │ ★任意多个订阅者★      │
  │ 耦合          │ 和 app 强绑定       │ ★发送方不知道接收方★  │
  │ 典型用途      │ ★鉴权/事务/改响应头★ │ ★日志/监控/埋点/审计★ │
  └──────────────┴────────────────────┴──────────────────────┘

★ ★同一件事的两种写法（★对照理解★）★：
  # 需求：记录每个请求的耗时
  ★用钩子★：
    @app.before_request
    def start_timer(): g._t0 = time.perf_counter()
    @app.after_request
    def log_time(resp):
        current_app.logger.info("%.1fms", (time.perf_counter()-g._t0)*1000)
        return resp                       # ★★必须 return★★

  ★用信号★：
    @request_started.connect_via(app)
    def start_timer(sender, **extra): g._t0 = time.perf_counter()
    @request_finished.connect_via(app)
    def log_time(sender, response, **extra):
        current_app.logger.info("%.1fms", ...)
                                          # ★不用 return★

  ★ 这个场景两者都行；★但如果要改响应头就只能用钩子★

★ ★什么时候必须用钩子★：
  ✓ ★鉴权拦截★（return redirect 中断）
  ✓ ★维护模式★（return 503）
  ✓ ★统一加响应头★（安全头、CORS、请求 ID 回显）
  ✓ ★开启/提交事务★
  ✓ ★请求级资源的创建与释放★

★ ★什么时候更适合用信号★：
  ✓ ★可插拔的功能★（装了扩展就有，没装也不影响）
  ✓ ★一个事件多个独立反应★（下单 → 发邮件 + 加积分 + 清缓存 + 推消息）
  ✓ ★不该侵入业务代码的横切关注点★（审计、埋点）
  ✓ ★扩展给用户提供的钩子★（用户不改扩展源码就能挂逻辑）
  ✓ ★测试时的观测点★（template_rendered）

钩子和信号的本质区别是「在不在流程里」。钩子是请求处理链条的一环：before_request 返回非 None 会中断请求直接返回响应、after_request 接收 response 且可以修改；信号则是旁路的通知，返回值被完全忽略。用同一个「记录请求耗时」的需求对照写两遍就很清楚——这个场景两者都行，但如果要改响应头就只能用钩子。定位很清晰：必须用钩子的是鉴权拦截、维护模式、统一加响应头、事务管理、请求级资源管理；更适合信号的是可插拔的功能、一个事件触发多个独立反应、不该侵入业务代码的横切关注点、以及扩展提供给用户的扩展点。

二、blinker 的机制

★ 基本概念：
  from blinker import signal
  sig = signal("my-event")             # ★同名 signal() 返回同一个对象★
  sig.connect(receiver)                # 订阅
  sig.send(sender, **kwargs)           # ★发送（同步）★
  sig.disconnect(receiver)             # 取消订阅
  sig.receivers                        # 当前订阅者
  bool(sig.receivers)                  # ★★判断有没有人订阅（性能优化）★★

★ ★Namespace：避免全局命名冲突★：
  from blinker import Namespace
  _signals = Namespace()               # ★自己的命名空间★
  order_paid = _signals.signal("order-paid")
  ★ 直接用 blinker.signal("x") 是★全局的★
    → 两个库都用了 "saved" 这个名字就会串
  ★ Flask 自己也是这么做的：flask.signals 里有个 _signals Namespace

★ ★sender 的作用（★关键★）★：
  sig.connect(fn)                      # ★接收所有发送者的信号★
  sig.connect(fn, sender=app)          # ★★只接收 app 发出的★★
  sig.connect_via(app)(fn)             # 装饰器形式

  ★ 为什么要限定 sender：
    ① ★测试时会创建多个 app 实例★ → 不限定会互相触发
    ② ★同一进程跑多个应用★
    ③ ★语义更精确★
  ★ Flask 的内置信号 ★sender 都是 app 对象★

★ ★★弱引用陷阱（最容易踩）★★：
  blinker 默认 ★weak=True★，用 weakref 保存接收者
  → ★接收者被 GC 后订阅自动消失★（设计意图：避免内存泄漏）

  ✗ 危险写法：
    def register(app):
        def handler(sender, **kw): ...    # ★局部函数★
        my_signal.connect(handler)        # ★register 返回后 handler 被回收★
    → ★信号不再触发，且没有任何报错★

  ✗ 同样危险：
    my_signal.connect(lambda s, **kw: ...)      # ★lambda 立刻被回收★
    my_signal.connect(obj.method)               # ★obj 被回收时也失效★

  ✓ 三种正确做法：
    ① ★模块级函数★（最推荐，天然有强引用）
    ② ★weak=False★：my_signal.connect(handler, weak=False)
       （★注意：要自己负责 disconnect，否则内存泄漏★）
    ③ ★保存引用★：app.extensions["my_handlers"] = [handler]

★ ★send 的行为★：
  results = sig.send(sender, key=value)
  # ★返回 [(receiver, return_value), ...]★
  ★ 但 ★Flask 完全忽略这些返回值★
  ★ 调用是★同步、顺序★的：
    → ★接收者耗时 = 请求变慢★
    → ★接收者抛异常 = 向上传播到发送方★（blinker 不吞）
  ★ 执行顺序：★不保证★（不要依赖注册顺序）

★ ★性能：先判断有没有订阅者★：
  # Flask 源码里的模式
  if request_started.receivers:            # ★避免无谓的参数构造★
      request_started.send(app, _async_wrapper=...)
  ★ 自己发信号时也可以这么写（★参数构造昂贵时★）

★ ★异步接收者（blinker 1.7+ / Flask 2.3+）★：
  async def on_paid(sender, **kw): ...
  ★ Flask 会用 app.ensure_sync 包装
  ★ 但仍是★在请求线程里同步等待完成★ → 不是真正的"异步不阻塞"

blinker 的核心是「同名 signal() 返回同一个对象」，所以要用 Namespace 避免全局命名冲突（Flask 自己也是这么做的）。sender 参数很关键——限定 sender=app 能避免测试时多个 app 实例互相触发，Flask 的内置信号 sender 都是 app 对象。最容易踩的是弱引用陷阱：blinker 默认 weak=True，接收者被 GC 后订阅静默消失且不报错——所以局部函数、lambda、绑定方法都可能突然失效；三种正确做法是用模块级函数（最推荐）、传 weak=False（但要自己负责 disconnect）、或显式保存引用。send() 是同步顺序调用的：接收者耗时会拖慢请求、抛异常会向上传播、执行顺序不保证（不要依赖注册顺序）。性能上可以学 Flask 源码先判断 sig.receivers 再构造参数。

三、内置信号的实战用法

★ ★用法一：请求日志与监控（最常见）★
  import time
  from flask import g, request, current_app
  from flask import request_started, request_finished, got_request_exception

  def _started(sender, **extra):
      g._start = time.perf_counter()
      g.request_id = request.headers.get("X-Request-Id") or uuid4().hex

  def _finished(sender, response, **extra):
      dur_ms = (time.perf_counter() - g._start) * 1000
      current_app.logger.info(
          "method=%s path=%s status=%s dur=%.1fms rid=%s",
          request.method, request.path, response.status_code,
          dur_ms, g.request_id)
      # ★上报指标★
      metrics.histogram("http_duration_ms", dur_ms,
                        tags={"endpoint": request.endpoint or "unknown",
                              "status": response.status_code})

  def _exception(sender, exception, **extra):
      current_app.logger.exception("unhandled rid=%s", g.get("request_id"))
      sentry_sdk.capture_exception(exception)

  def init_app(app):
      request_started.connect(_started, app)
      request_finished.connect(_finished, app)
      got_request_exception.connect(_exception, app)

★ ★got_request_exception 的重要细节★：
  - ★只在"未被 errorhandler 处理"的异常时触发★？
    ★不是★——它在异常传给 errorhandler ★之前★ 触发
    → ★即使你注册了 errorhandler，信号也会触发★
    → ★这正是 Sentry 能捕获所有异常的原因★
  - ★abort(404) 这类 HTTPException ★不会★触发它★
    （它们是正常的流程控制，不是"错误"）
  - ★debug 模式下也会触发★

★ ★用法二：模板测试（★官方推荐★）★
  from flask import template_rendered
  from contextlib import contextmanager

  @contextmanager
  def captured_templates(app):
      recorded = []
      def record(sender, template, context, **extra):
          recorded.append((template, context))
      template_rendered.connect(record, app)
      try:
          yield recorded
      finally:
          template_rendered.disconnect(record, app)

  def test_list_page(app, client):
      with captured_templates(app) as templates:
          client.get("/posts")
          assert len(templates) == 1
          tmpl, ctx = templates[0]
          assert tmpl.name == "posts/list.html"
          assert len(ctx["posts"]) == 3        # ★★断言上下文而不是 HTML★★
  ★ ★比 assert b"标题" in resp.data 健壮得多★
    （改个 CSS class 不会让测试挂掉）

★ ★用法三：appcontext_tearing_down 清理资源★
  from flask import appcontext_tearing_down
  def _teardown(sender, exc=None, **extra):
      conn = g.pop("my_conn", None)
      if conn: conn.close()
  appcontext_tearing_down.connect(_teardown, app)
  ★ 和 app.teardown_appcontext 钩子等价，★但信号版本更适合扩展★
    （扩展不需要独占那个钩子）

★ ★用法四：appcontext_pushed 做测试打桩★
  from flask import appcontext_pushed, g
  @contextmanager
  def user_set(app, user):
      def handler(sender, **kwargs):
          g.user = user                 # ★★在上下文推入时注入★★
      with appcontext_pushed.connected_to(handler, app):
          yield
  # 测试里：
  with user_set(app, admin_user):
      assert client.get("/admin").status_code == 200
  ★ ★官方文档推荐的测试技巧★

★ ★用法五：message_flashed 记录提示★
  from flask import message_flashed
  @message_flashed.connect_via(app)
  def log_flash(sender, message, category, **extra):
      current_app.logger.debug("flash [%s] %s", category, message)

内置信号最常见的用法是请求日志与监控——request_started 记起点、request_finished 记录状态码和耗时、got_request_exception 上报 Sentry。got_request_exception 有个重要细节：它在异常传给 errorhandler 之前触发，所以即使你注册了错误处理器，信号也照样触发——这正是 Sentry 能捕获全部异常的原因；而 abort(404) 这类 HTTPException 不会触发它（它们是正常的流程控制）。template_rendered 是官方推荐的模板测试方式——断言模板名和上下文数据，比断言 HTML 字符串健壮得多（改个 CSS class 不会让测试挂）。appcontext_pushed 配合 connected_to 做测试打桩也是官方文档推荐的技巧，能在上下文推入时注入 g.user。

四、自定义信号与业务解耦

★ 典型场景：下单成功后要做很多事
  ✗ ★耦合的写法★：
    def pay_order(order):
        order.status = "paid"
        db.session.commit()
        send_receipt_email(order)         # ★邮件模块★
        add_user_points(order.user)       # ★积分模块★
        clear_order_cache(order.user_id)  # ★缓存模块★
        notify_warehouse(order)           # ★仓储模块★
        update_statistics(order)          # ★统计模块★
    ★ 问题：
      ① ★pay_order 依赖 5 个模块★（改任何一个都要动它）
      ② ★加新功能要改核心业务代码★
      ③ ★测试 pay_order 要 mock 5 个东西★
      ④ ★一个非核心功能报错会让下单失败★

  ✓ ★信号解耦★：
    order_paid = _signals.signal("order-paid")
    def pay_order(order):
        order.status = "paid"
        db.session.commit()
        order_paid.send(current_app._get_current_object(), order=order)
    # ★各模块自己订阅，pay_order 一无所知★

★ ★但要清醒认识信号的代价（★面试加分点★）★：
  ✗ ★流程变得"隐式"★：
    看 pay_order 的代码★完全不知道会发生什么★
    → 新人排查问题时找不到"邮件是哪来的"
  ✗ ★调试困难★：断点打在哪？调用栈里全是 blinker
  ✗ ★顺序不保证★：不能依赖"先加积分再发邮件"
  ✗ ★异常影响发送方★：一个接收者炸了，后面的不执行，且异常向上抛
  ✗ ★事务边界模糊★：send 在 commit 之后，接收者里再写库就是新事务
  ★ ★所以：不是所有解耦都该用信号★

★ ★什么时候用信号，什么时候直接调用★：
  ┌────────────────────────────┬────────────────────────────┐
  │ ★用信号★                    │ ★直接调用★                  │
  ├────────────────────────────┼────────────────────────────┤
  │ 反应是★可选的/可插拔的★      │ ★核心业务流程★              │
  │ 反应★彼此独立★              │ 有★顺序依赖★                │
  │ ★允许失败★（埋点、通知）     │ ★不允许失败★（扣库存）      │
  │ ★扩展/插件提供的能力★        │ 本模块自己的逻辑            │
  │ 反应数量★会增长★            │ 固定的两三步                │
  └────────────────────────────┴────────────────────────────┘
  ★ ★关键判断：这件事失败了，主流程该不该失败？★
    → ★该失败 = 直接调用（并放进同一事务）★
    → ★不该失败 = 信号 + 接收者内部 try/except★

★ ★接收者的健壮写法（★必须★）★：
  @order_paid.connect
  def send_receipt(sender, order, **extra):
      try:
          send_email_task.delay(order.id)      # ★★投递到 Celery，不阻塞★★
      except Exception:
          current_app.logger.exception("发送收据失败 order=%s", order.id)
          # ★★吞掉异常，不影响主流程和其他接收者★★

★ ★参数约定★：
  def receiver(sender, **extra):       # ★★永远收 **extra★★
  ★ 原因：
    ① ★信号以后加参数时不会破坏已有接收者★
    ② Flask 内部会传一些额外的键
  ✗ def receiver(sender, order):       # ★信号加个参数就 TypeError★

★ ★不能传 LocalProxy 作为 sender★：
  ✗ order_paid.send(current_app, order=o)
    → ★current_app 是 LocalProxy，弱引用会出问题★
  ✓ order_paid.send(current_app._get_current_object(), order=o)

自定义信号的价值是业务解耦——pay_order 只管改状态和发信号，邮件、积分、缓存、仓储各自订阅，核心业务代码不依赖它们。但要清醒认识信号的代价，这是面试加分点：流程变得隐式（看代码完全不知道会发生什么，新人排查时找不到「邮件是哪来的」）、调试困难、顺序不保证、异常会影响发送方、事务边界模糊。所以关键判断是「这件事失败了，主流程该不该失败？」——该失败就直接调用并放进同一事务，不该失败才用信号，而且接收者内部必须 try/except 兜住异常、耗时操作要投递到 Celery。两个参数约定：接收函数永远要收 **extra（信号以后加参数时不破坏已有接收者），send() 的 sender 不能传 LocalProxy，要用 current_app._get_current_object()。

五、和其他机制的对比

★ Flask 信号 vs Django 信号：
  ┌────────────────┬──────────────────────────────────────┐
  │ ★Django★        │ ★内置大量模型层信号★                  │
  │                 │ pre_save/post_save/pre_delete/        │
  │                 │ m2m_changed/post_migrate...           │
  │                 │ → ★ORM 深度集成★                      │
  ├────────────────┼──────────────────────────────────────┤
  │ ★Flask★         │ ★只有请求/模板/上下文层面的信号★       │
  │                 │ ★没有模型信号★（模型是 SQLAlchemy 的事）│
  └────────────────┴──────────────────────────────────────┘
  ★ 想要模型层信号 → 用 ★SQLAlchemy 的 event 系统★：
    from sqlalchemy import event
    @event.listens_for(User, "after_insert")
    def on_user_created(mapper, connection, target): ...
    ★ 注意：★after_insert 里不能用 session★（用 connection）
    ★ 需要在 commit 后做事：监听 Session 的 "after_commit"

★ ★信号 vs 消息队列（★架构层面的对比★）★：
  ┌──────────────┬────────────────────┬────────────────────┐
  │              │ ★blinker 信号★      │ ★消息队列（MQ）★    │
  ├──────────────┼────────────────────┼────────────────────┤
  │ 范围          │ ★进程内★            │ ★跨进程/跨服务★     │
  │ 同步/异步     │ ★同步阻塞★          │ ★异步★             │
  │ 可靠性        │ ★进程崩了就丢★      │ ★持久化 + 重试★     │
  │ 顺序          │ 不保证              │ 可配置              │
  │ 成本          │ ★零★                │ 要部署和运维        │
  └──────────────┴────────────────────┴────────────────────┘
  ★ ★常见组合：信号做进程内分发，接收者里投递到 MQ★
    @order_paid.connect
    def fanout(sender, order, **kw):
        publish_to_mq("order.paid", {"id": order.id})

★ ★信号 vs 显式的事件总线★：
  自己实现一个 EventBus 也很简单（几十行）
  ★ 用 blinker 的好处：
    ① ★和 Flask 内置信号是同一套机制★（认知一致）
    ② ★sender 过滤、weak 引用、Namespace 都现成★
    ③ ★扩展生态在用★（Flask-Login 的 user_logged_in 等）
  ★ 自己实现的好处：★可以加类型提示、异步、优先级、错误隔离★

★ ★扩展提供的信号（了解一下）★：
  Flask-Login：user_logged_in / user_logged_out / user_login_confirmed
  Flask-Principal：identity_changed / identity_loaded
  Flask-SQLAlchemy：models_committed（★2.x 有，3.0 移除了★）
  ★ 这也是扩展"让用户挂逻辑而不改源码"的标准做法

★ ★Flask 3.0 的变化（★重要★）★：
  ┌────────────────────────────────────────────────────┐
  │ ★blinker 从"可选依赖"变成"必需依赖"★                  │
  │ → 3.0 之前：★没装 blinker 时信号静默失效★             │
  │   （signals_available = False，connect 变成空操作）   │
  │ → ★这曾是个隐蔽的坑★：代码写了信号，生产没装 blinker， │
  │   ★监控和审计悄悄全部失效★                            │
  │ → 3.0 起：★blinker 一定存在，signals_available 移除★   │
  └────────────────────────────────────────────────────┘

Flask 信号和 Django 信号的最大区别是：Django 内置了大量模型层信号（post_save、pre_delete），而 Flask 只有请求/模板/上下文层面的信号——因为模型是 SQLAlchemy 的事。想要模型层信号要用 SQLAlchemy 的 event 系统（注意 after_insert 里不能用 session、要在 commit 后做事得监听 after_commit）。信号和消息队列是不同层次的东西：信号是进程内、同步、进程崩了就丢，MQ 是跨服务、异步、持久化可重试——常见组合是「信号做进程内分发，接收者里投递到 MQ」。最后一个重要变化：Flask 3.0 起 blinker 从可选依赖变成必需依赖——3.0 之前没装 blinker 时信号会静默失效（signals_available = False），这曾是个隐蔽的坑：代码写了监控和审计，生产环境却因为没装 blinker 而全部失效。

六、实践清单

★ 检查清单：
  □ ★分清钩子和信号：要控制流程用钩子★
  □ ★connect 时限定 sender=app★
  □ ★接收者用模块级函数（防弱引用回收）★
  □ ★接收函数签名带 **extra★
  □ ★接收者内部 try/except 吞掉异常★
  □ ★耗时操作投递到 Celery，别在接收者里同步做★
  □ ★send 的 sender 用 _get_current_object()★
  □ ★自定义信号用 Namespace★
  □ ★不依赖接收者的执行顺序★
  □ ★核心业务不要用信号（用直接调用 + 同一事务）★
  □ ★临时订阅记得 disconnect（测试里用 contextmanager）★

★ 排查「信号没触发」的顺序：
  ① ★确认 connect 时的 sender 和 send 时的一致★
  ② ★确认接收者没被 GC★（是不是局部函数/lambda？）
  ③ ★print(sig.receivers) 看有没有订阅上★
  ④ ★确认注册代码真的被执行了★（模块被 import 了吗）
  ⑤ Flask < 3.0：★确认装了 blinker★

★ 一个完整的可观测性模块（实战模板）：
  # observability.py
  import time, uuid
  from flask import g, request, current_app
  from flask import request_started, request_finished, got_request_exception

  def _on_started(sender, **extra):
      g._t0 = time.perf_counter()
      g.request_id = request.headers.get("X-Request-Id", uuid.uuid4().hex)

  def _on_finished(sender, response, **extra):
      dur = (time.perf_counter() - g._t0) * 1000
      response.headers["X-Request-Id"] = g.request_id      # ★★这行要用钩子！★★
      current_app.logger.info(...)

  def _on_exception(sender, exception, **extra):
      current_app.logger.exception("rid=%s", g.get("request_id"))

  def init_app(app):
      request_started.connect(_on_started, app)
      request_finished.connect(_on_finished, app)
      got_request_exception.connect(_on_exception, app)
      # ★注意：改响应头必须用 after_request 钩子，信号里改不一定生效★
      @app.after_request
      def _echo_rid(resp):
          resp.headers["X-Request-Id"] = g.get("request_id", "")
          return resp

★ 一句话总结：
  ★"信号是『通知』——返回值被忽略、改不了流程、发送方不知道接收方；
    钩子是『流程的一环』——能中断请求、能改响应。
    要控制用钩子，要观察用信号；
    用信号记得限定 sender、用模块级函数防弱引用回收、
    接收者内部吞异常。"★

检查清单里最关键的四条：限定 sender=app、用模块级函数防弱引用回收、接收函数签名带 **extra、接收者内部 try/except。排查「信号没触发」有固定顺序：先确认 connect 和 send 的 sender 一致、再看接收者是不是被 GC 了、然后 print(sig.receivers)、确认注册代码被执行了（Flask 3.0 之前还要确认装了 blinker）。实战模板里有个容易忽略的细节：request_finished 信号触发时响应已经生成，在里面改响应头不一定能生效——要改响应必须用 after_request 钩子，这正好又印证了「信号改不了流程」这个核心区别。

记忆钩子：「Flask 的信号基于 ★blinker★，是发布-订阅机制——★发送方完全不知道接收方的存在★。★和 before_request 这类钩子最本质的区别是『能不能影响流程』★：★钩子是请求处理链条的一环★（before_request ★返回非 None 会中断请求直接返回响应★、after_request ★接收 response 且必须返回、可以修改★），★而信号是纯粹的通知——接收者的返回值被完全忽略，既不能中断流程也不能改写响应★。所以定位是：★要控制流程用钩子（鉴权拦截/维护模式/加响应头/事务），要旁路观察用信号（日志/监控/埋点/审计/缓存失效）★。★三个使用要点★：★① connect 时限定 sender=app★（否则测试里多个 app 实例会互相触发，Flask 内置信号的 sender 都是 app 对象）；★② blinker 默认用弱引用（weak=True）★——★局部函数/lambda/绑定方法被 GC 后订阅静默消失且不报错★，所以★用模块级函数★（最推荐）或传 weak=False；★③ send() 是同步顺序调用的★——★接收者耗时会拖慢请求、抛异常会向上传播影响发送方、执行顺序不保证★，所以★接收者内部必须 try/except、耗时操作投递 Celery★。接收函数★签名永远带 **extra★（信号加参数时不破坏已有接收者），★send 的 sender 不能传 LocalProxy，要用 current_app._get_current_object()★。内置信号重点：★got_request_exception 在异常传给 errorhandler 之前触发★（★所以注册了 errorhandler 它照样触发，这正是 Sentry 能捕获全部异常的原因★），★但 abort(404) 这类 HTTPException 不会触发它★；★template_rendered 是官方推荐的模板测试方式★（断言模板名和上下文，比断言 HTML 健壮）；★appcontext_pushed 配合 connected_to 做测试打桩注入 g.user★。自定义信号用 ★Namespace 避免全局命名冲突★。★信号的代价要清醒认识（面试加分点）★：★流程变隐式（看代码不知道会发生什么）、调试困难、顺序不保证、异常影响发送方、事务边界模糊★ → ★关键判断：这件事失败了主流程该不该失败？该失败就直接调用并放同一事务，不该失败才用信号★。★Flask 和 Django 的区别：Django 有大量模型层信号（post_save 等），Flask 没有★——模型层要用 ★SQLAlchemy 的 event 系统★。★Flask 3.0 起 blinker 从可选变成必需依赖★——★3.0 之前没装 blinker 时信号会静默失效★，曾让监控和审计悄悄全挂。」

七、常见误区与追问

误区：信号和 before_request 差不多，都能用来做权限拦截。 信号做不了拦截。before_request 的返回值有特殊语义——返回非 None 时 Flask 会跳过视图函数，直接把这个返回值当作响应，所以能写 return redirect("/login") 实现鉴权、return "维护中", 503 实现维护模式。而信号接收者的返回值虽然会被 send() 收集成 [(receiver, result), ...] 列表，但 Flask 拿到之后直接丢弃——你在 request_started 的接收者里 return redirect(...)，什么都不会发生，请求照常往下走。同理 after_request 能改响应头（因为它接收 response 并返回修改后的对象），而 request_finished 信号触发时响应已经定型，在里面改可能不生效。记住定位：钩子在流程里、能控制；信号在旁路、只能观察。
误区：signal.connect(handler) 之后订阅就一直有效。 blinker 默认用弱引用（weak=True），这意味着它不会阻止 handler 被垃圾回收——一旦没有其他强引用指向这个函数，订阅就静默消失了。最典型的踩坑写法是把接收者定义成局部函数：def register(app): def handler(...): ...; my_signal.connect(handler)——register() 返回后 handler 就没人引用了，下次 GC 一跑订阅就没了，而且不会有任何报错或警告。同样危险的还有 connect(lambda s, **kw: ...)（lambda 立刻被回收）和 connect(obj.method)（obj 被回收时失效）。这个设计的本意是避免内存泄漏（订阅者对象能被正常回收），但对使用者很不直观。三种正确做法：① 用模块级函数（天然有强引用，最推荐）；② 传 weak=False（但你要自己负责在合适的时候 disconnect）；③ 把 handler 存到某个长期存在的容器里（比如 app.extensions）。
误区：用信号解耦业务逻辑总是好的设计。 解耦是有代价的，而且代价往往被低估。最大的问题是流程变得隐式：一个新人打开 pay_order() 看到的只有「改状态 + commit + send 信号」，他完全不知道这一步之后会发邮件、加积分、清缓存、通知仓库——排查「为什么用户收到了两封邮件」时得先知道 blinker 的存在、再去全局搜谁订阅了这个信号。其次是调试困难（调用栈里全是 blinker 的转发）、顺序不保证（不能依赖「先加积分再发邮件」）、异常会向上传播（一个接收者炸了，后面的不执行、异常还抛回给了发送方）、事务边界模糊（send 在 commit 之后，接收者里再写库就是另一个事务，失败了主流程也回滚不了）。所以判断标准是一句话：「这件事失败了，主流程该不该失败？」——该失败的（扣库存、生成订单号）就直接调用并放进同一个事务；不该失败的（埋点、通知、缓存失效）才用信号，并且接收者内部必须自己吞掉异常。
误区：got_request_exception 只在没有 errorhandler 处理时才触发。 恰恰相反——它在异常被传给 errorhandler 之前就触发了。Flask 的处理流程是：视图抛出异常 → 发送 got_request_exception 信号 → 查找匹配的 errorhandler → 生成错误响应。所以即使你注册了 @app.errorhandler(Exception) 把所有异常都转成了漂亮的 JSON，信号依然会触发——这正是 Sentry 这类监控工具能捕获全部异常的原因（否则一旦项目加了全局错误处理器，监控就瞎了）。另一个容易搞混的点：abort(404)、abort(400) 这类 HTTPException 不会触发这个信号，因为它们是正常的流程控制而不是「程序出错」——如果 404 也上报 Sentry，告警会被淹没。想统计 4xx 就用 request_finished 看 response.status_code。
误区：Flask 的信号能像 Django 那样监听模型的保存和删除。 Flask 没有任何模型层信号——它的内置信号只覆盖请求生命周期、模板渲染、上下文进出、flash 消息这几个框架层面的时刻。原因很简单：Flask 本身不带 ORM，模型是 SQLAlchemy 的领域。要做「用户创建后自动建档案」「文章保存后更新搜索索引」这类事情，应该用 SQLAlchemy 自己的 event 系统：@event.listens_for(User, "after_insert")、"before_update"、"after_delete" 等。用它有两个必须知道的细节：① after_insert/after_update 这些 mapper 级事件里不能使用 session（正处于 flush 过程中，会破坏状态），只能用传入的 connection 执行原生 SQL；② 要在事务提交后才做的事（发邮件、清缓存、投消息队列）应该监听 Session 的 after_commit 事件，否则事务回滚了副作用却已经发生。顺带一提，Flask-SQLAlchemy 2.x 曾提供 models_committed 信号，但 3.0 已经移除了。
追问：为什么 signal.send() 的 sender 不能直接传 current_app？ 因为 current_app 是一个 LocalProxy 对象——它本身不是应用实例，只是一个转发器，每次访问属性时才去当前上下文里找真正的 app。这带来两个问题：① sender 匹配失效——接收者注册时写的是 connect(handler, app)（真实的 app 对象），而你发送时传的是 proxy，blinker 比较 sender 身份时两者不是同一个对象，于是限定了 sender 的接收者根本收不到信号；② 弱引用问题——blinker 内部会对 sender 做弱引用处理，而 LocalProxy 是个临时对象，行为不可预期。正确写法是 current_app._get_current_object()，它返回被代理的真实 app 实例。同样的道理适用于把 current_app、request、g 传给 Celery 任务、存进闭包、或者跨线程传递时——只要对象要「活得比当前上下文长」，就必须先 _get_current_object()。这也是 Flask 里 LocalProxy 相关的通用注意事项。
追问：信号的接收函数为什么一定要写 **extra？ 两个理由。① 向前兼容——Flask 或你自己的信号将来可能会增加新的关键字参数（比如 request_finished 某个版本多传了一个 duration），如果你的接收函数签名是 def on_finished(sender, response):，新参数一加就会 TypeError: got an unexpected keyword argument，而且是在运行时才炸。写成 def on_finished(sender, response, **extra): 就能安全忽略任何新增参数。② Flask 内部确实会传额外的键——比如某些版本会传 _async_wrapper 之类的内部参数。这是 blinker/Flask 官方文档明确建议的写法，可以理解成一条硬性约定：信号接收函数的签名永远是 (sender, ..., **extra)。顺带一提，参数名要和信号发送时的关键字对得上（request_finished 传的是 response=，你就得写 response），写错名字的话那个参数会掉进 **extra 里而你的形参拿不到值——这也是「信号触发了但接收者里的值是 None」的常见原因。
追问：什么时候该用信号，什么时候该直接上消息队列？ 两者解决的是不同层次的问题。blinker 信号是进程内的、同步的——send() 会在当前线程里顺序执行所有接收者，然后才返回；进程崩溃或重启，正在处理的通知就丢了，没有重试、没有持久化。它的优势是零成本（不用部署任何东西）、延迟极低、代码简单。消息队列（RabbitMQ/Kafka/Redis Stream）是跨进程甚至跨服务的、异步的——消息持久化、失败可重试、消费者可以独立扩容、发送方立刻返回不受消费方影响。判断标准：① 反应必须可靠送达（订单支付后必须发货）→ MQ；② 反应耗时较长（发邮件、生成 PDF、调外部 API）→ MQ 或 Celery；③ 反应是纯内存操作且允许丢失（清本进程缓存、更新计数器、埋点）→ 信号；④ 需要跨服务通知 → 一定是 MQ。实践中最常见的其实是组合使用：用信号做进程内的事件分发（解耦业务代码），在接收者里把消息投递到 MQ 或 Celery——这样业务代码只发一个信号，而可靠性和异步性由 MQ 保证。

八、加强记忆

Flask 的信号基于 blinker，是一套发布-订阅机制——发送方完全不知道接收方的存在。和 before_request 这类钩子最本质的区别是「能不能影响流程」：钩子是请求处理链条的一环（before_request 返回非 None 会中断请求直接返回响应，after_request 接收 response 且必须返回、可以修改），而信号是纯粹的通知——接收者的返回值被完全忽略，既不能中断流程也不能改写响应。所以定位很清晰：要控制流程用钩子（鉴权拦截、维护模式、加响应头、事务管理），要旁路观察用信号（日志、监控、埋点、审计、缓存失效）。三个使用要点：① connect 时限定 sender=app（否则测试里多个 app 实例会互相触发，Flask 内置信号的 sender 都是 app 对象）；② blinker 默认用弱引用（weak=True）——局部函数、lambda、绑定方法被 GC 后订阅会静默消失且不报错，所以要用模块级函数（最推荐）或传 weak=False；③ send() 是同步顺序调用的——接收者耗时会拖慢请求、抛异常会向上传播影响发送方、执行顺序不保证，所以接收者内部必须 try/except 兜住、耗时操作要投递到 Celery。接收函数签名永远带 **extra（信号将来加参数时不会破坏已有接收者），send() 的 sender 不能传 LocalProxy，要用 current_app._get_current_object()。内置信号的重点：got_request_exception 在异常传给 errorhandler 之前触发（所以注册了错误处理器它照样触发，这正是 Sentry 能捕获全部异常的原因），但 abort(404) 这类 HTTPException 不会触发它；template_rendered 是官方推荐的模板测试方式（断言模板名和上下文，比断言 HTML 健壮得多）；appcontext_pushed 配合 connected_to 做测试打桩注入 g.user。自定义信号要用 Namespace 避免全局命名冲突。信号的代价必须清醒认识：流程变得隐式（看代码不知道会发生什么）、调试困难、顺序不保证、异常影响发送方、事务边界模糊——关键判断是「这件事失败了，主流程该不该失败？」，该失败就直接调用并放进同一事务，不该失败才用信号。Flask 和 Django 的一个重要区别：Django 有大量模型层信号（post_save 等），Flask 完全没有——模型层要用 SQLAlchemy 的 event 系统（注意 after_insert 里不能用 session，要在提交后做事得监听 after_commit）。最后，Flask 3.0 起 blinker 从可选依赖变成了必需依赖——3.0 之前没装 blinker 时信号会静默失效，曾让不少项目的监控和审计悄悄全挂。

简化版

详细版