JAVA 获取Canal请求到的MySQL数据库BINLOG日志数据

绝地灬酷狼 2023-02-13 14:40 4阅读 0赞

# 一、介绍： #

本文章是通过java项目连接Canal客户端获取，Canal通过模拟从库dump下来的BINLOG二进制文件数据。

# 二、引入pom文件： #

<dependency>
                <groupId>com.alibaba.otter</groupId>
                <artifactId>canal.client</artifactId>
                <version>1.1.1</version>
     </dependency>

# 三、Test方法测试 #

import com.alibaba.otter.canal.client.CanalConnector;
    import com.alibaba.otter.canal.client.CanalConnectors;
    import com.alibaba.otter.canal.protocol.CanalEntry;
    import com.alibaba.otter.canal.protocol.Message;
    import com.google.common.collect.ImmutableMap;
    import com.google.common.collect.Lists;
    import com.google.common.collect.Ordering;
    import org.apache.commons.codec.digest.DigestUtils;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.http.HttpEntity;
    import org.springframework.http.HttpHeaders;
    import org.springframework.http.HttpMethod;
    import org.springframework.http.ResponseEntity;
    import org.springframework.http.client.SimpleClientHttpRequestFactory;
    import org.springframework.web.client.RestTemplate;
    
    import java.net.InetSocketAddress;
    import java.nio.charset.Charset;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    import java.util.stream.Collectors;
    
    @SpringBootApplication
    public class TestCanal {
    
        @org.junit.Test
        public void getEntries() {
            CanalConnector connector = CanalConnectors
                    .newSingleConnector(new InetSocketAddress("10.179.251.***", 11111), "example", "", "");//写入正确的ip和端口以及destination
            int batchSize = 1000;
            connector.connect();//获取连接
            connector.subscribe(".*\\..*");//通配此处不做更改
            connector.rollback();
            try {
                while (true) {
                    Message message = connector.getWithoutAck(batchSize);//获取指定数量的数据
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {//当message.getId不等于-1或0时代表监控到了数据
                    } else {
                        printEntry(message.getEntries());//具体分析数据类，可根据自己需求进行修改
                    }
                    connector.ack(batchId); // 提交确认
                }
            } finally {
                connector.disconnect();
            }
        }
    
        private void printEntry(List<CanalEntry.Entry> list) {
            for (CanalEntry.Entry entry : list) {
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                        || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                    continue;//如果是事务跳过
                }
                CanalEntry.RowChange rowChage = null;
                try {
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());//获取修改的数据row
                } catch (Exception e) {
                }
                CanalEntry.EventType eventType = rowChage.getEventType();//获取类型 增删改
                if (entry.getHeader().getSchemaName().equals("mdata_dashboard")) {//库名
                    if (entry.getHeader().getTableName().equals("import_second_request_volume")) {//表名
                        System.out.println("changed.");
                        //取值根据表数据自行定义修改
                        List<String> deleteUserId = new ArrayList<String>();
                        List<String> updateUserId = new ArrayList<String>();
                        for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                            if (eventType == eventType.DELETE) {
                                deleteUserId.add(rowData.getBeforeColumnsList().get(1).getValue());
                            } else if (eventType == eventType.INSERT || eventType == eventType.UPDATE) {
                                updateUserId.add(rowData.getAfterColumnsList().get(1).getValue());
                                updateUserId.add(rowData.getBeforeColumnsList().get(1).getValue());
                            }
                        }
                    }
                }
            }
        }
    }

如果只想获取字段数据

/**
         * 处理数据
         *
         * @param list
         */
        private void printEntry(List<Entry> list, long batchId) {
            CryptCenterHandlerInvokeProxy proxy = ApplicationContextCapture
                    .getBean(CryptCenterHandlerInvokeProxy.class);
            for (Entry entry : list) {
                if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
                        || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                    continue;//跳过事务
                }
                RowChange rowChage = null;
                try {
                    rowChage = RowChange.parseFrom(entry.getStoreValue());//获取变更的row数据
                } catch (Exception e) {
                }
                EventType eventType = rowChage.getEventType();//获取变动类型
                entry.getHeader().getSchemaName();//获取变动库名
                entry.getHeader().getTableName();//获取变动表名
    
                for (RowData rowData : rowChage.getRowDatasList()) {
                    Map<String, Object> columnMap = new HashMap<String, Object>();
                    if (eventType == EventType.DELETE) {
                        columnMap = printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == EventType.INSERT) {
                        columnMap = printColumn(rowData.getAfterColumnsList());
                    } else {
                        columnMap = printColumn(rowData.getAfterColumnsList());
                    }
                    
                }
    
            }
        }
    
    /**
    *获取map 字段名，字段值， column.getUpdated()是否被更新等等，按需要进行取值
    */
    private Map<String, Object> printColumn(List<CanalEntry.Column> columns) {
            Map<String, Object> map = new HashMap<>();
    
            for (CanalEntry.Column column : columns) {
                map.put(column.getName(), column.getValue());
            }
            return map;
        }

打印columnMap数据结果

{
    	  interface_uri = /gateway/hr - basic / openApi / V2 / getStaffInfoByLdaps,
    	  total = 844147,
    	  day_at = 2,
    	  hour_at = 18,
    	  time_at = 20200602185630,
    	  second_at = 30,
    	  id = 10341,
    	  minute_at = 56,
    	  app_id = 181211329,
    	  month_at = 202006
    }

参考文章：[https://www.jianshu.com/p/23f5db8b3e25?mType=Group&from=androidqq][https_www.jianshu.com_p_23f5db8b3e25_mType_Group_from_androidqq]

[https_www.jianshu.com_p_23f5db8b3e25_mType_Group_from_androidqq]: https://www.jianshu.com/p/23f5db8b3e25?mType=Group&from=androidqq