MySQL触发器+ZeroMQ：不写业务代码也能实现数据库变更消息推送

简介数据库进程间通信是高并发与实时系统中的关键议题传统定时轮询效率低且浪费资源。这份技术方案以消息队列MQ为桥梁针对数据库无法直接向外部进程传递信号的问题给出了一套兼顾同步与异步模式的完整思路尤其适合无法改动既有代码或外包项目需要二次扩展的场景。文档从背景与应用场景切入列举了通过MQ发送短信邮件、触发图片处理、实时校验身份证、网站动态页面静态化及跨库数据同步等典型需求并说明如何用SELECT zmq_client(...)将任务发送到服务端。后半部分聚焦MySQL插件不仅讲解了zmq_client UDF函数的用法还给出了UDF插件的开发、编译、装载步骤以及基于AFTER INSERT/UPDATE/DELETE触发器自动向MQ推送消息的配置示例让数据库变更真正驱动外部流程。包内为一份116KB的docx文档共1个文件由浅入深适合直接对照实践。目前已有98人学习可供数据库工程师、架构师及需要实现进程间异步协作的开发者参考。1. 数据库进程间通信不用轮询让触发器把消息推给 MQ接手一个外包交付的电商项目最怕的就是“动代码”。商品表改个价格要顺藤摸瓜检查七八处写操作稍有不慎就把下单流程改出问题。这份《数据库进程间通信解决方案之 MQ》给出的思路很直接不改业务代码在 MySQL 里装一个 UDF 插件建几个触发器当数据发生变化时把主键、手机号、图片路径这些消息通过 MQ 推给外部进程由外部程序去发短信、处理图片、刷新静态页面。整个过程绕开了业务代码也绕开了“每隔几秒轮询一次数据库”的老办法。适合手里有存量 MySQL 项目、又不敢轻易改代码或者需要在数据库变动时触发外部操作的从业者。2. 为什么是 MQ从 FIFO 到 ZeroMQ 的进程间通信选型2.1 定时轮询的两个副作用延迟与无效查询很多人第一反应是写个循环定时查表看有没有新增或变更的数据。常见做法是在程序里起一个后台线程每 5 秒执行一次查询把新拿到的行交给业务逻辑处理。思路没错但代价有两个。第一是延迟不可控。5 秒轮询一次数据变更后平均要等 2.5 秒才能被发现。业务上如果希望“数据一变就触发”这个延迟就是硬伤。把轮询间隔缩到 1 秒延迟是降下来了但查询频率上去了数据库压力直接翻倍。第二是无效查询太多。绝大多数轮询周期里表根本没有新数据但查询照样全表扫或走索引区间扫描。数据量小的时候无感一旦表上了百万行轮询本身就是一种持续的负载。下面是典型的轮询伪代码while True: rows query(SELECT * FROM demo WHERE updated_at %s, last_seen) for row in rows: process(row) last_seen now() time.sleep(5)这段代码把“检测变化”和“处理变化”耦合在同一个进程里数据库还承担了持续查询的压力。进程一重启last_seen 丢失要么重复处理要么漏处理。用 MQ 方案替代轮询数据库只在数据真正变化时发一条消息没有变化就没有查询也没有轮询延迟。2.2 从 FIFO 到 ZeroMQ跨越操作系统边界方案作者早期的做法是 FIFO 管道也就是 Linux 命名管道。数据变更时把消息写进管道文件外部进程从管道另一端读取。FIFO 本身是可靠的进程间通信IPC机制但它有明确边界通信双方必须在同一台操作系统内。文档原文写得清楚FIFO 属于 IPC 机制限定在同一个 OS 或服务器内。后来作者用 ZeroMQ 重新实现升级成了 RPC 形态的通信同时兼容了跨越操作系统的场景。这里要区分两个概念IPC 是进程间通信机制解决的是本机进程之间怎么交换数据RPC 是远程过程调用模型解决的是“像调用本地函数一样调用远端能力”。UDF 里调用zmq_client(tcp://server001:5555, mobile)语法上就是一个函数调用实际消息却通过网络被路由到另一台机器上的服务端这就是从 IPC 走到 RPC 的关键变化。2.3 inproc / tcp / ipc三种协议怎么选ZeroMQ 支持三种传输协议文档里给出了对应的端点写法。选型时先想清楚“通信双方在哪里”再决定用哪种协议典型端点适用位置说明inprocinproc://my_publisher同一进程内的线程间线程间通信ITC不走网络栈延迟最低ipcipc:///tmp/feeds/0同一台主机上的进程间走 Unix socket避免占用 TCP 端口本机专用tcptcp://server001:5555跨主机、跨服务器最常用MySQL 与 MQ 服务端可以分开部署我的选择习惯是MQ 服务端和 MySQL 在同一台机器优先用ipc省掉 TCP 端口监听也没有防火墙问题MQ 服务端在独立服务器必须用tcpinproc一般出现在插件内部或同一个进程内的多线程通信直接写在 SQL 里意义不大。文档里的示例统一用tcp://localhost:5555是为了测试方便生产环境建议按上面这张表重新判断。3. MySQL 插件四个 UDF 的编译、装载与验证3.1 插件是什么UDF 与触发器的配合MySQL 的 UDFUser Defined Function是用户自定义函数编译成动态库.so文件放在 MySQL 插件目录里装载以后就能像内置函数一样在 SQL 语句中调用。它和存储过程不同UDF 更接近CONCAT()、NOW()这类标量函数可以直接嵌在 SELECT、触发器、EVENT 里返回一个值给调用方。文档说明共提供了 4 个 UDF公开给出装载示例的有两个zmq_client(socket, message)负责把消息发送到指定端点返回成功或失败标识zmq_publish用于发布模式适合一对多的消息广播。另外两个函数没有在文档里给出装载示例以源码仓库实现为准。这套设计最大的价值在于业务代码不用改数据库触发器里直接调用 UDF数据一变消息就出去了。3.2 编译环境用 apt 装齐三个依赖包编译前先把环境装齐。文档给的是 Debian/Ubuntu 系的命令核心依赖有三个pkg-config用于编译时定位头文件和库路径libmysqlclient-dev提供 MySQL 客户端开发库编译工具链负责把源码做成.so。sudo apt-get install pkg-config sudo apt-get install libmysqlclient-dev sudo apt-get install gcc g make cmake注意最后一个包名文档里写的是gcc gcc-c make cmake但gcc-c是 CentOS/RHEL 系的写法Debian/Ubuntu 下要装g。我一般会先跑一次mysql_config --version确认客户端开发库版本和运行中的 MySQL 大版本一致避免编译产物和服务器不匹配。3.3 cmake 编译与 so 文件部署源码来自 GitHub 上的mysql-zmq-plugin仓库克隆下来以后用 cmake 生成构建文件make 编译make install 安装。完整流程如下git clone https://github.com/netkiller/mysql-zmq-plugin.git cd mysql-zmq-plugin cmake . make make installcmake 阶段会探测 MySQL 头文件路径和 ZeroMQ 库是否存在输出一段检测结果如果提示找不到 libzmq说明系统里还没有安装 ZeroMQ 运行时库得先apt-get install libzmq-dev。make install 完成后动态库会被复制到插件目录但不同发行版的目标路径不完全一样建议手动确认一下ls -l /usr/lib/mysql/plugin/libzeromq.so如果没有生成在这个路径先查 MySQL 实际使用的插件目录再复制过去。查询方法在下一节。3.4 装载与卸载create function 和 mysql.func 表动态库放好之后用CREATE FUNCTION把函数注册到 MySQL。这条语句不需要重启 MySQL执行完立即生效CREATE FUNCTION zmq_client RETURNS STRING SONAME libzeromq.so; CREATE FUNCTION zmq_publish RETURNS STRING SONAME libzeromq.so;RETURNS STRING声明的是 UDF 返回值类型SONAME指定动态库文件名MySQL 会去插件目录里找这个.so。卸载时用DROP FUNCTION同样不需要重启DROP FUNCTION IF EXISTS zmq_client; DROP FUNCTION IF EXISTS zmq_publish;装载成功后MySQL 会在系统库的mysql.func表里留下记录这是一张真实存在的系统表记录了所有已注册的 UDF。查询方式SELECT * FROM mysql.func WHERE name LIKE zmq%;返回结果类似这样nameretdltypezmq_client0libzeromq.sofunctionzmq_publish0libzeromq.sofunctionret为 0 表示返回值类型是字符串编码dl是动态库文件名type为function表示 UDF 类型。这张表在排查问题时非常有用后面避坑章节会再提到。4. 触发器 zmq_client 实战短信、JSON 与页面静态化4.1 最小可运行验证先起服务端再发消息编译完插件先不要急着建触发器按最小闭环验证一遍启动仓库test目录下的服务端程序再让 MySQL 客户端发一条消息。cd test cmake . make ./server./server启动后会监听 5555 端口等待消息。此时另开一个终端进入 MySQLSELECT zmq_client(tcp://localhost:5555, Hello world!);执行结果--------------------------------------------------- | zmq_client(tcp://localhost:5555, Hello world!) | --------------------------------------------------- | Hello world! OK | ---------------------------------------------------返回内容由服务端程序决定Hello world!是消息原样回显OK是服务端处理完成后的反馈。回头再看服务端终端应该已经打印了Received: Hello world!。这条链路通了说明 UDF 能发、服务端能收后面接触发器才有意义。4.2 触发器落地AFTER INSERT / UPDATE / DELETE 三件套最小验证通过后把zmq_client调用写进触发器。最常见的做法是给业务表建三个触发器分别对应 INSERT、UPDATE、DELETE。以demo表为例字段是id、name、mobileDELIMITER $$ CREATE DEFINERdba% TRIGGER demo_after_insert AFTER INSERT ON demo FOR EACH ROW BEGIN SELECT zmq_client(tcp://localhost:5555, NEW.id); END$$ DELIMITER ;三个容易忽略的点。第一DELIMITER必须改因为BEGIN...END内部的分号会被 MySQL 客户端当成语句结束符不改会语法报错。第二DEFINER指定触发器以dba身份执行这个账号需要有调用zmq_client的权限这里直接用文档里的写法。第三INSERT 和 UPDATE 触发器里用NEW.id读取新行的主键DELETE 触发器要改成OLD.id。CREATE DEFINERdba% TRIGGER demo_after_update AFTER UPDATE ON demo FOR EACH ROW BEGIN SELECT zmq_client(tcp://localhost:5555, NEW.id); END; CREATE DEFINERdba% TRIGGER demo_after_delete AFTER DELETE ON demo FOR EACH ROW BEGIN SELECT zmq_client(tcp://localhost:5555, OLD.id); END;执行完后插一条数据测试MQ 服务端会立刻收到消息。注意触发器里SELECT zmq_client(...)的返回值不会显示但这个调用是真实执行的服务端输出就是最直接的证据。4.3 静态化案例从动态 URL 到文件覆盖文档里电商静态化的案例很有代表性。商品页是动态 URLhttp://www.example.com/goods.php?id100需要把它变成静态文件100.html。正常情况下得在后台所有商品写操作后面加缓存更新逻辑外包项目里只能从外部处理。三个触发器把商品 ID 推给 MQ 服务端后服务端程序做三件事收到 ID 就下载动态 URL 的内容保存成静态 HTMLINSERT 消息触发的动作是新建页面UPDATE 消息覆盖原有页面DELETE 消息删除页面。这样无论商品价格、属性怎么变静态化程序都会跟着响应而商品模块的代码一行没动。服务端接收消息的最小骨架常见做法是下面这样import zmq ctx zmq.Context() sock ctx.socket(zmq.REP) sock.bind(tcp://*:5555) while True: goods_id sock.recv_string() page fetch(http://www.example.com/goods.php?id%s % goods_id) save(/data/html/%s.html % goods_id, page) sock.send_string(OK)这段代码里的fetch和save换成实际的 HTTP 下载与文件写入即可。REP模式要求服务端收到消息后必须回一条消息客户端zmq_client才会返回“OK”。真实环境里下载失败的页面要做重试队列不能简单丢弃。4.4 同步还是异步按任务性质选文档里多个案例背后其实都藏着一个选型问题同步调用还是异步调用。发短信和邮件建议异步手机号码是否有效无从确认到达率本身不是 100%消息进队列就让程序自行处理把成功或失败记入日志就行。图片处理也建议异步处理耗时可能达到秒级同步等结果会把数据库连接拖住。身份证号码校验则适合同步。校验逻辑在 MQ 端几乎不耗时数据库这边调用zmq_client等一小会儿就能拿到结果直接反馈给调用方。选型标准我是这样定的任务本身耗时超过网络往返时间就异步业务必须在当前事务里拿到处理结果才同步。还有一点要提醒触发器里同步调用zmq_client会阻塞当前 INSERT 或 UPDATE 语句如果 MQ 服务端处理慢整个写操作都会变慢这也是文档反复建议异步的根本原因。5. 避坑指南消息发不出去、函数重复装载与并发边界5.1 zmq_client 返回失败服务端收不到消息现象在 MySQL 客户端执行SELECT zmq_client(tcp://localhost:5555, Hello)返回的不是OK而是失败标识或空结果服务端终端也没有任何输出。原因最常见的三个MQ 服务端没启动端口写错客户端连 5555服务端监听 6666服务端的 socket 类型和客户端不匹配客户端发的请求服务端不是用对等的 pattern 接收。文档里zmq_client的返回值描述为“成功返回 true失败返回 false”但实际返回内容取决于服务端程序怎么写不能只看函数名。解决先起服务端再做客户端测试。确认服务端监听正常后用nc -vz localhost 5555检查端口连通性再回到 MySQL 执行一条最简单的消息。如果还不行看 MySQL 错误日志UDF 调用失败通常会在 error log 里留下线索。5.2 create function 报 Function already exists现象执行CREATE FUNCTION zmq_client ...时报错错误码是ERROR 1125 (HY000): Function zmq_client already exists。原因这台 MySQL 之前已经装载过同名 UDFmysql.func表里残留了记录或者你在测试环境装过一次后来忘了卸载直接在另一套环境里重复执行。解决先查后建。执行SELECT * FROM mysql.func WHERE name LIKE zmq%;确认现状。如果存在记录先DROP FUNCTION zmq_client;再重新创建。最稳的做法是把装载和卸载写成一个脚本对每次部署都先卸载再装载幂等执行。5.3 插件目录能找到 so 文件装载还是报错现象ls /usr/lib/mysql/plugin/能看到libzeromq.so但CREATE FUNCTION报ERROR 1126 (HY000): Cant open shared library libzeromq.so。原因MySQL 的插件目录不一定是/usr/lib/mysql/plugin/。make install把文件装到了系统的默认路径但 MySQL 实际读取的plugin_dir指向别处另一种可能是 so 文件本身依赖的 libzmq 运行时库不在动态链接器的搜索路径里。解决先执行SELECT plugin_dir;拿到真实目录把 so 文件复制过去chmod 755保证可读。然后跑ldd /usr/lib/mysql/plugin/libzeromq.so看输出里有没有not found。如果有说明 libzmq 没装全apt-get install libzmq-dev后重试。5.4 事务里调用 UDF消息比预期多且收不回来现象一个事务里 UPDATE 了三行数据MQ 服务端收到了三条消息。事务最终 ROLLBACK 了但消息已经发出去了事务回滚没能“撤销”消息。原因UDF 调用是在语句执行时实时发生的MySQL 不会把 UDF 发送消息的行为纳入事务控制。触发器里的AFTER INSERT、AFTER UPDATE也是在每一行操作之后立刻执行与最终是否 COMMIT 无关。这套方案的语义是“至少一次投递”不是“精确一次”。解决接受这个语义MQ 端做幂等处理。以静态化为例同一个商品 ID 的多次更新都指向同一个静态文件路径重复覆盖不会有副作用。如果业务上必须事务提交成功后才通知外部就不能用触发器直发得改成事务提交后写一张变更表由外部程序或 EVENT 定时扫表发送。5.5 apt 下安装编译依赖报包名不存在现象执行sudo apt-get install gcc-c提示Unable to locate package gcc-c。原因文档里的依赖清单写的是gcc gcc-c make cmake其中gcc-c是 CentOS/RHEL 系的 C 编译器包名。Debian/Ubuntu 系的包名是g照抄文档命令在 Ubuntu 上必然装不上。解决换成sudo apt-get install gcc g make cmake pkg-config。顺带提醒如果你在 CentOS 上编译libmysqlclient-dev也要对应换成mysql-community-devel发行版不同包名不能直接搬。5.6 触发器里发送长消息返回失败但服务端已处理现象用CONCAT拼 JSON 字符串发送消息长度到几百字节时zmq_client偶尔返回失败但服务端其实已经收到了消息。原因ZeroMQ 的 TCP 消息上限默认是很大但 MySQL 客户端与服务端之间的交互模式、以及服务端是否回包都会影响客户端感知。更常见的是 JSON 拼接出错字段里有单引号或特殊字符没转义SQL 语句本身就把数据截断了。解决拼消息时先做一次转义。文档里的 JSON 示例是CONCAT({name:, name, , tel:, mobile, })这在字段值没有引号的场景下能跑真实数据里 name 出现单引号就直接炸。我一般建议改用CONCAT拼 URL 编码或 Base64或者用JSON_OBJECT函数生成标准 JSON避免手工拼串引入格式问题。6. 端到端验证链路先裸测服务端再上触发器6.1 上线前强制走一遍的验证清单UDF 插件和触发器都建好之后要用一条完整的验证链路确认每个环节都正常而不是直接对生产表建触发器。我习惯按下面的顺序过一遍验证项执行方式期望结果确认插件目录SELECT plugin_dir;输出目录路径so 文件已复制到该目录裸测服务端cd test ./server服务端正常监听 5555 端口客户端裸测SELECT zmq_client(tcp://localhost:5555, smoke);返回携带OK服务端打印Received: smoke检查装载记录SELECT * FROM mysql.func WHERE name LIKE zmq%;能看到 zmq_client、zmq_publish 记录触发器回归向测试表插入一行数据服务端打印出该行主键再执行 UPDATE、DELETE 确认对应动作6.2 什么时候不该用这套方案这套方案的适用边界要划清楚。它适合存量 MySQL 项目、不适合动业务代码的场景也适合数据变更频率不高、对“至少一次投递”语义可接受的系统。但如果你的数据变更量极大需要精确捕获每一行变更并同步到数仓更成熟的替代方案是订阅 binlog比如 Canal 或 Debezium它们连 DDL 都能抓到。如果你用的是云数据库没有自定义插件权限UDF 这条路根本走不通只能退回到变更表加外部扫表的方案。做过一次生产环境的上线踩过 5.2 的坑开发环境的mysql.func表里已经残留了同名记录生产环境直接create function报 already exists还以为动态库没装对。从那以后我每次上线都强制走一遍上面的验证清单确认初始状态干净再动手。希望帮到你。本文还有配套的精品资源点击获取