写在家远程办公的第三周，手机里的消息有点紧张，生怕漏掉一条有用的信息，不可能是这样的，公司濒临崩溃，大家都如履薄冰，毕竟这时候失业有点惨（穷怕了）。但就工作而言，还是比较清闲的。比起在公司工作，我的空闲时间多了很多，可以随心所欲地做一些自己喜欢的事情。而且我发现，一个人一旦闲下来，真的很可怕，潜在的人才会被全面爆发。我闺蜜是抖音的深度患者，一年不做一次饭，每周做两次饭给我吃。一顿黑饭，烤馒头版的“饼”，糊版的“凉皮”，然后就和厕所结下了不解之缘。..不过，作为一个程序员，我对暗黑料理并不是很感兴趣。我还是喜欢在业余时间研究和钻研一些新技术。运河成了很好的研究对象。无意中监控了公司MySQL的一举一动首先，什么是canal？Canal是阿里开发的基于数据库增量日志解析，提供增量数据订阅和消费的框架。整个框架纯JAVA开发。目前只支持Mysql和MariaDB（类似于mysql）。那么什么是数据库增量日志呢？MySQL日志的种类很多，主要有：错误日志、查询日志、慢查询日志、事务日志、二进制日志。MySQL数据库中发生的数据变化（DML（datamanipulationlanguage）数据操作语言，也就是我们熟悉的增删改查）都会以二进制日志（binarylog）的形式存储。2.canal的原理在介绍canal的原理之前，我们先来回顾一下MySQL主从同步的原理，或许可以让你更好的理解canal的工作机制。1、MySQL主从同步原理：MySQL主从同步也叫读写分离，可以提高数据库的负载和容错能力，实现数据库的高可用。如有侵权，联系删除主节点操作流程：当主节点数据发生变化（删除、更新、插入，或创建函数、存储过程等）时，将记录写入二进制日志。这些记录也称为二进制日志事件（binarylogevents）。showbinlogevents这些事件会按顺序写入bin日志。当从节点开始连接主节点时，主节点会为从节点启动binlogdump线程（负责传输binlog数据）。一旦主节点的binlog发生变化，binlogdump线程会通知从节点有可以传输的binlog，并将相应的binlog内容发送给从节点。从节点运行过程：从节点上会创建两个线程：一个I/O线程和一个SQL线程。I/O线程连接到master节点，master节点上的binlogdump线程会将binlog的内容发送给I\O线程。I/O线程收到binlog内容后，将内容写入本地relaylog。sql线程读取I/O线程写入的relaylog，并将relaylog中的内容写入从库。2.Canal原理了解了上面MySQL的主从同步原理后，canal的工作机制就很容易理解了。实际上，canal模拟了MySQL数据库中从节点和主节点的交互协议，伪装成MySQL从节点，向MySQL主节点发送dump协议。MySQL主节点收到转储请求后，开始将二进制日志推送到从节点（也就是canal）。以上图片来自网络。如有侵权，请联系删除。光说不练假把戏，动手吧！3、Canal实现“监控”MySQL。在写代码之前，我们先修改MySQL。安装MySQL我就不细说了，只说基本操作。1、查看MySQL是否开启了二进制日志功能showbinarylogs。如果没有启用，就是图中的状态。一般用户没有这个命令权限，我却有，tsktsktsk！如果不需要手动开启，在my.cnf文件中配置binlog-format为Row模式log-bin=mysq-binbinlog-format=Rowlog-bin为binlog文件存放位置binlog-format设置MySQL复制方式log-binMySQL三种复制方式：基于SQL语句的复制（statement-basedreplication，SBR）优点：修改数据的sql保存在binlog中，不需要记录每条sql和数据变化，binlog体积会小，IO开销会少，性能好缺点：会造成master-slaveRow-basedreplication(RBR)的数据不一致优点：不记录每条SQL语句的上下文信息，只需要记录修改了哪些数据，修改为缺点：binlog非常大，尤其是使用altertable属性时，会产生大量的binlog数据mixed-basedreplication(MBR)。对应的，binlog有三种格式：STATEMENT、ROW、MIXED。2.为canal创建有操作MySQL权限的用户CREATEUSERcanalIDENTIFIEDBY'canal';GRANTSELECT,REPLICATIONSLAVE,REPLICATIONCLIENTON*.*TO'canal'@'%';--GRANTALLPRIVILEGESON*.*TO'canal'@'%';FLUSHPRIVILEGES;3.安装canal下载地址：https://github.com/alibaba/canal/releases下载后选择版本，例如：canal.deployer-xxx.tar.gz4.配置canal，修改instance.properties文件，需要添加监控数据库和表规则，canal可以全量监控数据库，也可以监控某个表，更加灵活。vimconf/example/instance.properties#####################################################mysqlserverIdcanal.instance.mysql.slaveId=2020#positioninfo修改自己的数据库（canal要监控的数据库地址）canal.instance.master.address=127.0.0.1:3306canal.instance.master.journal.name=canal.instance.master.position=canal.instance.master.timestamp=#canal.instance.standby.address=#canal.instance.standby.journal.name=#canal.instance.standby。position=#canal.instance.standby.timestamp=#username/password改成自己数据库信息的账号（准备阶段单独开一个账号）canal.instance.dbUsername=canalcanal.instance.dbPassword=canalcanal。instance.defaultDatabaseName=canal.instance。connectionCharset=UTF-8#tableregex表监控规则#canal.instance.filter.regex=blogs\.blog_infocanal.instance.filter.regex=.\*\\\\..\*#tableblackregexcanal.instance.filter.black。regex=启动canalshbin/startup.sh，查看服务器日志，确认canal是否正常启动。vilogs/canal/canal.log显示现在canal服务器运行成功。2020-01-0815:25:33.361[主要]INFOcom。alibaba.otter.canal.deployer.CanalLaunc她-##startthecanalserver.2020-01-0815:25:33.468[main]INFOcom.alibaba.otter.canal.deployer.CanalController-##startthecanalserver[192.168.12.245:11111]2020-01-0815:25:34.061[main]INFOcom.alibaba.otter.canal.deployer.CanalLauncher-##thecanalserverisrunningnow......5.编写Java客户端代码，实现canal监控，导入依赖包com.alibaba.ottercanal.client1.1.0这里只是简单实现了publicclassMainApp{publicstaticvoidmain(String...args)throwsException{/***creation和*/CanalConnectorconnector=CanalConnectors.newSingleConnector(newInetSocketAddress(AddressUtils.getHostIp(),11111),"example","","");intbatchSize=1000;intemptyCount=0;try{connector.connect();/***监控数据库*/connector.subscribe(".*\\..*");/***指定要监控的表，库名.表名*///connector.subscribe("xin-master.jk_order");connector.rollback();//不是d检测到120次心跳后跳出inttotalEmptyCount=120;while(emptyCountentrys){for(CanalEntry.Entryentry:entrys){if(entry.getEntryType()==CanalEntry.EntryType.TRANSACTIONBEGIN||entry.getEntryType()==CanalEntry.EntryType.TRANSACTIONEND){继续;}CanalEntry.RowChangerowChage=空；尝试{rowChage=CanalEntry.RowChange.parseFrom(entry.getStoreValue());}catch(Exceptione){thrownewRuntimeException("错误##parseroferomanga-eventhasanerror,数据:"+entry.toString(),e);}CanalEntry.EventTypeeventType=rowChage.getEventType();System.out.println(String.format("================>binlog[%s:%s],name[%s,%s],eventType:%s",entry.getHeader().getLogfileName(),entry.getHeader().getLogfileOffset(),entry.getHeader().getSchemaName(),entry.getHeader().getTableName(),eventType));for(CanalEntry.RowDatarowData:rowChage.getRowDatasList()){if(eventType==CanalEntry.EventType.DELETE){printColumn(rowData.getBeforeColumnsList());}elseif(eventType==CanalEntry.EventType.INSERT){printColumn(rowData.getAfterColumnsList());}else{System.out.println("------>之前");printColumn(rowData.getBeforeColumnsList());System.out.println("-------->after");printColumn(rowData.getAfterColumnsList());}}}}privatestaticvoidprintColumn(Listcolumns){for(CanalEntry.Columncolumn:columns){System.out.println(column.getName()+":"+column.getValue()+"update="+column.getUpdated());}}}代码写到此为止。我们启动服务看看效果如何。由于没有操作数据库，所以监控结果都是空的。接下来，我们在数据库中执行一条更新语句，尝试updatejk_ordersetorder_no='1111'whereid=40。控制台检测数据库的修改，生成binlog日志文件mysql-bin.000009:3830。那么如何使用和分析生成的binlog文件中的SQl语句呢？com.github.shyikomysql-binlog-connector-java0.13.0下载刚才的binlog文件，在本地测试publicstaticvoidmain(String[]args)throwsIOException{StringfilePath="C:\\ProgramData\\MySQL\\MySQLServer5.7\\Data\\mysql-bin.000009:3830";FilebinlogFile=newFile（文件路径）；EventDeserializerreventDeserializer=newEventDeserializer（）；eventDeserializer.setChecksumType（ChecksumType.CRC32）；BinaryLogFileReaderreader=newBinaryLogFileReader（binlogFile，eventDeserializer）；try{for（Eventevent；（event=reader.readEvent=null；））！){System.out.println(event.toString());}}finally{reader.close();}}查看执行结果，发现最近对数据库的操作是添加一个idx_index索引Event{header=EventHeaderV4{timestamp=1551325542000,eventTtype=ANONYMOUS_GTID，serverId=1，headerLength=19，dataLength=46，nextPosition=8455，flags=0}，data=null}Event{header=EventHeaderV4{timestamp=1551325542000，eventType=QUERY，serverId=1，headerLength=19,dataLength=190,nextPosition=8664,flags=0},data=QueryEventData{threadId=25,executionTime=0,errorCode=0,database='xin-master',sql='ALTERTABLE`jk_order`DROPINDEX`idx_index`,ADDINDEX`idx_index`(`user_id`,`service_id`,`real_price`)USINGBTREE'}}Event{header=EventHeaderV4{timestamp=1551438586000,eventType=STOP,serverId=1,headerLength=19,dataLength=4,nextPosition=8687,flags=0},data=null}至此我们就实现了对MySQL的监控。4、Canal应用场景Canal应用场景大致如下：解决MySQL主从同步延迟问题，实现数据库实时备份。多级索引（卖家和买家各分库索引）实现业务缓存刷新价格变化等重要业务消息重点分析canal是如何解决mysql主从同步延迟问题对于部署在机房的集群，有将是一个同步延迟。举个栗子：一个订单状态为未支付，主节点修改为已支付。但是由于某些原因，延迟的数据不能及时同步到slave。这时候用户可以马上查看订单的状态（通过slave查询），显示支付还在pending中。哪位用户看到这种情况不要惊慌。为什么会有主从同步延迟？当主库master的TPS并发高时，master节点并发产生的修改操作，而slave节点的SQL线程是单线程处理同步数据，自然会出现延迟。然而，主从同步的原因不止这些。由于主从服务器是跨机器、跨机房存在的，所以除了网络带宽，网络的稳定性和机器之间的同步性是造成主从同步的主要原因。综上所述，本文只是对渠道监控数据库功能的简单实现。它旨在为您提供解决问题的方法。重复的还是那句话。解决问题的技术方法是非常正确的。如何应用需要结合具体业务。
