万字分享，我是如何一步一步监控公司MySQL的？
发布时间：2020-08-14 14:55:21 所属栏目：编程 来源：网络整理
导读：写在前边 在家远程办公第三周，快被手机上的消息搞的有些神经质了，生怕错过一条有用的信息，没办法形势如此，公司摇摇欲坠大家也都如履薄冰，毕竟这时候失业有点惨（ 穷怕了 ）。 但就干活来说还是比较清闲的，和在公司上班相比，清闲下来很多碎片时间，
|
显示canal server is running now即为成功 2020-01-08 15:25:33.361 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.2020-01-08 15:25:33.468 [main] INFO com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[192.168.12.245:11111]2020-01-08 15:25:34.061 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......5、编写Java客户端代码，实现canal监听 引入依赖包 <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.0</version></dependency>这里只是简单实现 public class MainApp { public static void main(String... args) throws Exception { /** * 创建与 */ CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(), 11111), "example", "", ""); int batchSize = 1000; int emptyCount = 0; try { connector.connect(); /** * 监控数据库中所有表 */ connector.subscribe(".*..*"); /** * 指定要监控的表，库名.表名 */ //connector.subscribe("xin-master.jk_order"); connector.rollback(); //120次心跳过后未检测到，跳出 int totalEmptyCount = 120; while (emptyCount < totalEmptyCount) { Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据 long batchId = message.getId(); int size = message.getEntries().size(); if (batchId == -1 || size == 0) { emptyCount++; System.out.println("empty count : " + emptyCount); try { Thread.sleep(1000); } catch (InterruptedException e) { } } else { emptyCount = 0; // System.out.printf("message[batchId=%s,size=%s] n", batchId, size); printEntry(message.getEntries()); } /** * 提交确认 */ connector.ack(batchId); /** * 处理失败, 回滚数据 */ connector.rollback(batchId); } System.out.println("empty too many times, exit"); } finally { connector.disconnect(); /** * 手动开启事务回滚 */ //TransactionAspectSupport.currentTransactionStatus().setRollbackOnly(); } } private static void printEntry(List<CanalEntry.Entry> entrys) { for (CanalEntry.Entry entry : entrys) { if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry .EntryType .TRANSACTIONEND) { continue; } CanalEntry.RowChange rowChage = null; try { rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e); } CanalEntry.EventType eventType = 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.RowData rowData : rowChage.getRowDatasList()) { if (eventType == CanalEntry.EventType.DELETE) { printColumn(rowData.getBeforeColumnsList()); } else if (eventType == CanalEntry.EventType.INSERT) { printColumn(rowData.getAfterColumnsList()); } else { System.out.println("-------> before"); printColumn(rowData.getBeforeColumnsList()); System.out.println("-------> after"); printColumn(rowData.getAfterColumnsList()); } } } } private static void printColumn(List<CanalEntry.Column> columns) { for (CanalEntry.Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated()); } }}代码到这就编写完成了，我们启动服务看下是什么效果，由于并没有操作数据库，所以监听的结果都是空的。 （编辑：核心网） 【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容! |
