MySQL 系列：第11篇 触发器与事件调度器

IT策士 10余年一线大厂经验，专注 IT 思维、架构、职场进阶。我会在各个平台持续发布最新文章，助你少走弯路。

前面的存储过程和自定义函数让我们学会了如何“封装”逻辑供主动调用。但还有一种需求：当某件事发生时，自动执行一段逻辑——比如插入订单后自动更新库存、每天凌晨自动清理过期数据。这就是 MySQL 的触发器（Trigger）和事件调度器（Event Scheduler）。今天用 Python 配合实战，把它们彻底讲透，并警示你生产环境中的坑。

1. 准备数据：订单 + 库存 + 日志三张表

我们用“下单自动扣库存、记日志”这一经典场景来演示触发器。

importmysql.connector conn=mysql.connector.connect(host="127.0.0.1",port=3306,user="root",password="MyNewPass123!",database="shop")cursor=conn.cursor()# 商品表cursor.execute(""" CREATE TABLE IF NOT EXISTS products(idINT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(100)NOT NULL, stock INT NOT NULL DEFAULT0, sales INT NOT NULL DEFAULT0)ENGINE=InnoDB""")# 订单表cursor.execute(""" CREATE TABLE IF NOT EXISTS orders(idINT AUTO_INCREMENT PRIMARY KEY, product_id INT NOT NULL, quantity INT NOT NULL, total_price DECIMAL(10,2)NOT NULL, customer_name VARCHAR(50), status VARCHAR(20)DEFAULT'pending', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)ENGINE=InnoDB""")# 日志表cursor.execute(""" CREATE TABLE IF NOT EXISTS operation_log(idINT AUTO_INCREMENT PRIMARY KEY, table_name VARCHAR(50), operation VARCHAR(20), record_id INT, detail TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)ENGINE=InnoDB""")# 清空并灌入测试数据cursor.execute("TRUNCATE products")cursor.execute("TRUNCATE orders")cursor.execute("TRUNCATE operation_log")products=[(1,"机械键盘",100,0),(2,"蓝牙耳机",50,0),(3,"显示器",30,0),]cursor.executemany("INSERT INTO products (id, title, stock, sales) VALUES (%s,%s,%s,%s)", products)conn.commit()print("✅ 测试环境准备完毕")

2. 触发器（Trigger）：数据变更时的自动响应

触发器绑定在表上，在INSERT / UPDATE / DELETE之前或之后自动执行。MySQL 5.7 之前一张表同事件同时机只能有一个触发器，8.0 放宽此限制（可多个，按创建顺序执行）。

2.1 触发器六种时机

2.2 实战：订单创建后自动扣库存

create_trigger_sql=""" CREATE TRIGGER after_order_insert AFTER INSERT ON orders FOR EACH ROW BEGIN -- 扣减库存 UPDATE products SET stock=stock - NEW.quantity, sales=sales + NEW.quantity WHEREid=NEW.product_id;-- 记录日志 INSERT INTO operation_log(table_name, operation, record_id, detail)VALUES('orders','INSERT', NEW.id, CONCAT('客户 ', NEW.customer_name,' 购买了 product_id=', NEW.product_id,' x', NEW.quantity));END""" cursor.execute(create_trigger_sql)print("✅ 触发器 after_order_insert 创建成功")

核心关键字：

  • NEW：引用插入或更新后的新行（INSERT 和 UPDATE 中使用）

  • OLD：引用删除或更新前的旧行（DELETE 和 UPDATE 中使用）

2.3 用 Python 测试触发器

# 查询触发前状态cursor.execute("SELECT id, title, stock, sales FROM products WHERE id = 1")row=cursor.fetchone()print(f"📊 下单前 - {row[1]}: 库存={row[2]}, 销量={row[3]}")# 插入订单（触发器自动扣库存）cursor.execute(""" INSERT INTO orders(product_id, quantity, total_price, customer_name)VALUES(1,3,1197.00,'张三')""")conn.commit()print("✅ 订单插入成功（触发器已自动执行）")# 查询触发后状态cursor.execute("SELECT id, title, stock, sales FROM products WHERE id = 1")row=cursor.fetchone()print(f"📊 下单后 - {row[1]}: 库存={row[2]}, 销量={row[3]}")# 查看日志cursor.execute("SELECT * FROM operation_log ORDER BY id DESC LIMIT 1")log=cursor.fetchone()print(f"\n📋 自动日志: {log[4]} @ {log[5]}")

预期输出：

✅ 触发器 after_order_insert 创建成功 📊 下单前 - 机械键盘: 库存=100, 销量=0✅ 订单插入成功（触发器已自动执行） 📊 下单后 - 机械键盘: 库存=97, 销量=3📋 自动日志: 客户 张三 购买了product_id=1x3 @2025-07-2212:00:00

触发器的效果透明且强制——无论通过哪种方式插入订单（应用代码、命令行、其他存储过程），库存都会自动扣减。

2.4 BEFORE 触发器：数据校验

在插入前检查库存是否充足，如果不足则拒绝：

cursor.execute(""" CREATE TRIGGER before_order_insert BEFORE INSERT ON orders FOR EACH ROW BEGIN DECLARE current_stock INT;SELECT stock INTO current_stock FROM products WHEREid=NEW.product_id;IF current_stock IS NULL THEN SIGNAL SQLSTATE'45000'SET MESSAGE_TEXT='商品不存在';ELSEIF current_stock<NEW.quantity THEN SIGNAL SQLSTATE'45000'SET MESSAGE_TEXT='库存不足';END IF;END""")print("✅ BEFORE 触发器创建成功")# 测试库存不足try: cursor.execute("INSERT INTO orders (product_id, quantity, total_price, customer_name) VALUES (1, 999, 9999.00, '李四')")conn.commit()except mysql.connector.Error as e: print(f"❌ 触发器拦截: {e}")

预期输出：

✅ BEFORE 触发器创建成功 ❌ 触发器拦截:1644(45000): 库存不足

SIGNAL SQLSTATE '45000'是 MySQL 抛出自定义异常的标准方式，会让插入操作直接失败。

2.5 AFTER UPDATE 触发器：价格变动告警

当商品价格大幅波动时，自动记录告警日志：

# 先给 products 表加 price 列cursor.execute("ALTER TABLE products ADD COLUMN IF NOT EXISTS price DECIMAL(10,2) DEFAULT 0")cursor.execute(""" CREATE TRIGGER after_product_price_change AFTER UPDATE ON products FOR EACH ROW BEGIN IF NEW.price!=OLD.price THEN INSERT INTO operation_log(table_name, operation, record_id, detail)VALUES('products','UPDATE', NEW.id, CONCAT('价格变动: ', OLD.price,' -> ', NEW.price,' (变动幅度: ', ROUND((NEW.price-OLD.price)/OLD.price*100,1),'%)'));END IF;END""")print("✅ 价格监控触发器创建成功")# 测试改价cursor.execute("UPDATE products SET price = 399.00 WHERE id = 1")cursor.execute("UPDATE products SET price = 499.00 WHERE id = 1")conn.commit()cursor.execute("SELECT detail FROM operation_log WHERE operation = 'UPDATE' AND table_name = 'products'")print("\n📋 价格变动日志：")forrowincursor.fetchall(): print(f" {row[0]}")

预期输出：

✅ 价格监控触发器创建成功 📋 价格变动日志： 价格变动:0.00->399.00(变动幅度:0.0%)价格变动:399.00->499.00(变动幅度:25.1%)

注意OLD.price和NEW.price的对比：UPDATE 触发器中既能访问旧值也能访问新值。

2.6 查看与删除触发器

# 查看所有触发器cursor.execute("SHOW TRIGGERS LIKE 'orders'")print("\n📋 orders 表上的触发器：")forrowincursor.fetchall(): print(f" {row[0]} - {row[4]} {row[5]}")# 删除触发器cursor.execute("DROP TRIGGER IF EXISTS after_product_price_change")print("✅ 触发器已删除")

3. 事件调度器（Event Scheduler）：数据库里的 Cron

触发器是被动的（等数据变更），而事件调度器是主动的——它按照预设的时间计划自动执行任务，类似于 Linux 的crontab。

3.1 开启事件调度器

# 查看调度器状态cursor.execute("SHOW VARIABLES LIKE 'event_scheduler'")print(f"事件调度器状态: {cursor.fetchone()[1]}")# 开启（需要 SUPER 权限）cursor.execute("SET GLOBAL event_scheduler = ON")print("✅ 事件调度器已开启")

3.2 创建定时任务：每天清理过期日志

cursor.execute(""" CREATE EVENT IF NOT EXISTS clean_old_logs ON SCHEDULE EVERY1DAY STARTS CURRENT_TIMESTAMP DO DELETE FROM operation_log WHERE created_at<DATE_SUB(NOW(), INTERVAL90DAY)""")print("✅ 定时清理事件创建成功")

  • EVERY 1 DAY：每天执行一次

  • STARTS CURRENT_TIMESTAMP：从现在开始

  • ENDS可选，指定结束时间

3.3 创建复杂事件：每日销售统计

cursor.execute(""" CREATE TABLE IF NOT EXISTS daily_stats(dateDATE PRIMARY KEY, order_count INT DEFAULT0, total_revenue DECIMAL(12,2)DEFAULT0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)ENGINE=InnoDB""")cursor.execute(""" CREATE EVENT IF NOT EXISTS generate_daily_report ON SCHEDULE EVERY1DAY STARTS CONCAT(CURDATE(),' 23:59:00')DO BEGIN INSERT INTO daily_stats(date, order_count, total_revenue)SELECT CURDATE(), COUNT(*), COALESCE(SUM(total_price),0)FROM orders WHERE DATE(created_at)=CURDATE()ON DUPLICATE KEY UPDATE order_count=VALUES(order_count), total_revenue=VALUES(total_revenue);END""")print("✅ 每日报表事件创建成功")

3.4 用 Python 管理事件

# 查看所有事件cursor.execute("SHOW EVENTS FROM shop")print("\n📋 shop 数据库的事件：")forrowincursor.fetchall(): print(f" {row[1]} - 每 {row[3]} 执行 - 状态: {row[5]}")# 临时禁用事件cursor.execute("ALTER EVENT clean_old_logs DISABLE")print("✅ clean_old_logs 已禁用")# 手动触发一次事件（方便测试）cursor.execute("ALTER EVENT clean_old_logs ENABLE")cursor.execute("ALTER EVENT clean_old_logs ON COMPLETION PRESERVE")# 删除事件cursor.execute("DROP EVENT IF EXISTS generate_daily_report")print("✅ generate_daily_report 已删除")

4. 性能陷阱与替代方案

触发器和事件看似强大，但生产环境中经常被称为“隐式炸弹”：

4.1 触发器的致命缺陷

4.2 事件调度器的注意事项

  • 事件由单线程执行，一个事件阻塞会拖累其他事件

  • 大量事件或长事务事件会导致事件延迟

  • 事件执行失败不会自动重试，需要自行监控

4.3 替代方案

我的建议：99% 的触发器需求都可以用应用层逻辑替代。如果你正在设计一个新系统，优先在 Python 服务中显式实现业务逻辑，把 MySQL 当成纯粹的数据存储引擎。只有在遗留系统维护或者对数据库层的透明性有硬性要求时，才考虑触发器。

5. 动手试试：完善自动化体系

基于现有表结构，完成以下挑战：

  1. 创建一个 AFTER DELETE 触发器，当订单被删除时自动回退库存（stock + OLD.quantity, sales - OLD.quantity），并记录日志。

  2. 创建一个事件，每小时执行一次，将status='pending'且创建超过 24 小时的订单自动取消（设为status='expired'）。

  3. 用 Python 插入一条订单后删除它，验证库存是否回到原始值。

  4. 思考题：如果一条订单插入触发了 3 个 AFTER INSERT 触发器，它们的执行顺序是怎样的？如何指定？

参考代码：

# 1. AFTER DELETE 触发器cursor.execute(""" CREATE TRIGGER after_order_delete AFTER DELETE ON orders FOR EACH ROW BEGIN UPDATE products SET stock=stock + OLD.quantity, sales=sales - OLD.quantity WHEREid=OLD.product_id;INSERT INTO operation_log(table_name, operation, record_id, detail)VALUES('orders','DELETE', OLD.id, CONCAT('退回库存 x', OLD.quantity));END""")print("✅ 删除回退触发器创建成功")

6. 总结

今天我们掌握了 MySQL 的两种自动机制：

  • 触发器：数据变更时自动执行，适合审计日志、强制校验，但性能开销大、调试困难。

  • 事件调度器：定时执行任务，适合数据清理、报表生成，但不应替代专业调度系统。

  • 核心原则：能用应用层逻辑解决的问题，不要下沉到数据库。触发器是“隐式魔法”，在团队协作中往往带来更多麻烦。

下一篇我们将进入权限与安全管理，学习如何精细化控制数据库访问。下次见！

想了解更多还可以去各个平台搜索「IT策士」，一起升级 IT 思维 ！