本文共 3523 字，大约阅读时间需要 11 分钟。

最近在新公司搭了一套canal. 按照<<dynamicTopic问题>>设置了canal.mq.topic和canal.mq.dynamicTopic

意图将一些不符合dynamicTopic匹配的语句的消息发送到一个默认的topic而避免报错INVALID_TOPIC_EXCEPTION

# table regexcanal.instance.filter.regex=fanboshi\\..*,sysbench\\..*# table black regexcanal.instance.filter.black.regex=.*\\.\\_.*\\_ghc,.*\\.\\_.*\\_gho,.*\\.\\_.*\\_del# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch# mq configcanal.mq.topic=default_topic# dynamic topic route by schema or table regex#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*canal.mq.dynamicTopic=.*\\..*#canal.mq.partition=0# hash partition config#canal.mq.partitionsNum=3#canal.mq.partitionHash=test.table:id^name,.*\\..*

因为和前几天同事说之前使dynamicTopic遇到了bug. 今天正准备自己搞些语句测一测, 于是想看下default_topic能否消费到消息(如果消费到了就说明我执行的语句无法被dynamicTopic规则匹配到而发送到了这个"默认Topic")

结果发现大量消息

{"data":null,"database":"","es":1583581779000,"id":655469,"isDdl":false,"mysqlType":null,"old":null,"pkNames":null,"sql":"UPDATE sbtest1 SET c=? WHERE id=?","sqlType":null,"table":"sbtest1","ts":1583581779149,"type":"QUERY"}{"data":null,"database":"","es":1583581779000,"id":655469,"isDdl":false,"mysqlType":null,"old":null,"pkNames":null,"sql":"DELETE FROM sbtest8 WHERE id=?","sqlType":null,"table":"sbtest8","ts":1583581779149,"type":"QUERY"}{"data":null,"database":"","es":1583581779000,"id":655469,"isDdl":false,"mysqlType":null,"old":null,"pkNames":null,"sql":"INSERT INTO sbtest8 (id, k, c, pad) VALUES (?, ?, ?, ?)","sqlType":null,"table":"sbtest8","ts":1583581779149,"type":"QUERY"}{"data":null,"database":"","es":1583581779000,"id":655470,"isDdl":false,"mysqlType":null,"old":null,"pkNames":null,"sql":"insert into fanboshi.monitor_delay(ctime) values(now())","sqlType":null,"table":"monitor_delay","ts":1583581779465,"type":"QUERY"}{"data":null,"database":"","es":1583581780000,"id":655471,"isDdl":false,"mysqlType":null,"old":null,"pkNames":null,"sql":"insert into fanboshi.monitor_delay(ctime) values(now())","sqlType":null,"table":"monitor_delay","ts":1583581780570,"type":"QUERY"}{"data":null,"database":"","es":1583581781000,"id":655472,"isDdl":false,"mysqlType":null,"old":null,"pkNames":null,"sql":"insert into fanboshi.monitor_delay(ctime) values(now())","sqlType":null,"table":"monitor_delay","ts":1583581781475,"type":"QUERY"}

一开始有点懵, 后来细看发现这些居然都是"statement"格式的, 是原始语句

于是搜了下github

https://github.com/alibaba/canal/issues/1361

参考一下FAQ里的解释：https://github.com/alibaba/canal/wiki/FAQ

问1：INSERT/UPDATE/DELETE被解析为Query或DDL语句？

答1：

1. binlog格式为非row模式，通过show variables like 'binlog_format’可以查看. 针对statement/mixed模式，DML语句都会是以SQL语句存在
2. mysql5.6+之后，在binlog为row模式下，针对DML语句通过一个开关(binlog-rows-query-log-events=true, show variables里也可以看到该变量)，记录DML的原始SQL，对应binlog事件为RowsQueryLogEvent，同时也有对应的row记录. ps. canal可以通过properties设置来过滤：canal.instance.filter.query.dml = true

我这里肯定是row格式了. 那么可能就是binlog_rows_query_log_events = 1的原因, 看了下还真是设置成了on

至于作者说的

ps. canal可以通过properties设置来过滤：canal.instance.filter.query.dml = true

https://github.com/alibaba/canal/wiki/AdminGuide

这解释不是很清楚的样子, 到底是过滤啥啊, 不管了反正用不着

现在怀疑那次同事遇到问题是不是就是因为就是开启了binlog_rows_query_log_events

转载地址：http://pkvub.baihongyu.com/