mysql binlog 解析

刺骨的言语ヽ痛彻心扉 2021-10-19 18:48 1002阅读 0赞

## 开启、查看binlog ##

1、查看是否开启binlog  
show variables like ‘log\_bin%’;

2、查看binlog位置  
show variables like ‘%datadir%’;

3、查看binlog  
show binary logs;  
show binlog events in ‘binlog.000031’;

4、使用mysqlbinlog 查看binlog内容  
\#mysqlbinlog --base64-output=decode-rows -v --database=test D:\\dev\\mysql-8.0.15-winx64\\data\\binlog.000033

5、配置server-id  
6、配置需要监听的数据库  
7、为用户授权从库同步binlog的权限

## 使用shyiko的mysql-binlog-connector-java解析binlog数据 ##

本例使用shyiko的mysql-binlog-connector-java将数据监听到，并转化为可用的map，再将map转化为java model的完整实例，转化部分为可选

#### 监听binlog ####

@Component
    @Slf4j
    public class BinLogEventListener implements CommandLineRunner {
    
        @Autowired
        private BinlogProperties binlogProperties;
    
        @Autowired
        private BinaryLogDataListener binaryLogDataListener;
    
        @Async
        @Override
        public void run(String... args) throws Exception {
            // 创建binlog监听客户端
            BinaryLogClient client = new BinaryLogClient(binlogProperties.getHost(),
                    binlogProperties.getPort(), binlogProperties.getUser(), binlogProperties.getPassword());
            client.setServerId(binlogProperties.getServerId());
            client.registerEventListener(binaryLogDataListener);
            client.connect();
        }
    }

#### Binlog监听器 ####

@Component
    @Slf4j
    public class BinaryLogDataListener implements BinaryLogClient.EventListener {
    
        @Autowired
        private JdbcTemplate jdbcTemplate;
    
        @Autowired
        private Converter baseConverter;
    
        @Autowired
        private DataProceser dataProceser;
    
        @Autowired
        private RedisUtil redisUtil;
    
        private Map<String, List<String>> columnNamesMap = new HashMap<>();
        private HashMap<Long, TableData> tableMap = new HashMap<>();
    
        private static final String METHOD_GET_TABLE_ID_NAME = "getTableId";
    
        @Value("${binlog.event.listener.database}")
        private List<String> database;
    
        @Override
        public void onEvent(Event event) {
            EventData data = event.getData();
            EventHeader header = event.getHeader();
    
    
            try {
                EventType eventType = header.getEventType();
                // 表信息
                TableData table = null;
                if (eventType == TABLE_MAP) {
                    if (data instanceof TableMapEventData) {
                        TableMapEventData tableMapEventData = (TableMapEventData) data;
                        if (tableMapEventData != null) {
                            // 没有监听的库不处理
                            if (!database.contains(tableMapEventData.getDatabase())) {
                                return;
                            }
                            log.info(" 时间:" + Calendar.getInstance().getTime() + "tableId: " + tableMapEventData.getTableId() + ", tableName: " + tableMapEventData.getDatabase() + "." + tableMapEventData.getTable());
                            table = new TableData(tableMapEventData.getDatabase(), tableMapEventData.getTable());
                            tableMap.put(tableMapEventData.getTableId(), table);
                        }
                    }
                }
                //-------------------------------
                // 每次数据变动都会先走TABLE_MAP 事件，需要处理的表tableMap里肯定有，如果没有，则不需要监听处理该表
                try {
                    if (!tableMap.containsKey(data.getClass().getDeclaredMethod(METHOD_GET_TABLE_ID_NAME).invoke(data))) {
                        return;
                    }
                } catch (Exception e) {
                    log.debug("{}", e);
                    return;
                }
                String method = null;
                boolean isProcess = false;
                try {
                    // 最后的Map数据
                    List<Map<String, Object>> datas = null;
                    List list = null;
                    List<String> columnNames = null;
                    // 删除
                    if (eventType == DELETE_ROWS || eventType == EXT_DELETE_ROWS) {
                        isProcess = true;
                        DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
                        method = "DELETE";
                        if (data != null) {
                            table = tableMap.get(deleteRowsEventData.getTableId());
                            log.info(" 时间:{}，method:{}，tableName:{}，sqlParameter:{}", Calendar.getInstance().getTime(), method, table, JSON.toJSONString(Arrays.asList(deleteRowsEventData.getRows())));
                            columnNames = getColumnNames(table);
                        }
                    }
                    // 新增
                    if (eventType == WRITE_ROWS || eventType == EXT_WRITE_ROWS) {
                        isProcess = true;
                        WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
                        method = "INSERT";
                        if (data != null) {
                            table = tableMap.get(writeRowsEventData.getTableId());
                            log.info(" 时间:{}，method:{}，tableName:{}，sqlParameter:{}", Calendar.getInstance().getTime(), method, table, JSON.toJSONString(Arrays.asList(writeRowsEventData.getRows())));
                            columnNames = getColumnNames(table);
                        }
                    }
                    // 更新
                    if (eventType == UPDATE_ROWS || eventType == EXT_UPDATE_ROWS) {
                        isProcess = true;
                        method = "UPDATE";
                        if (data != null) {
                            UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
                            log.info(" 时间:{}，method:{}，tableName:{}，sqlParameter:{}", Calendar.getInstance().getTime(), method, table, JSON.toJSONString(Arrays.asList(updateRowsEventData.getRows())));
                            table = tableMap.get(updateRowsEventData.getTableId());
                            columnNames = getColumnNames(table);
                        }
                    }
    
                    log.debug("event------------------->" + header.getEventType());
                    log.info("表：{}，字段：{}", table, columnNames);
                    datas = rowsToMap(data, columnNames);
                    if (isProcess && null != datas) {
    
                        list = baseConverter.convert(new BinlogData<>(table.getDatabase(), table.getTable(), datas));
                        dataProceser.post(new BinlogData<>(table.getDatabase(), table.getTable(), datas), list);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } catch (Exception e) {
                //异常处理
                e.printStackTrace();
            } finally {
                //释放锁
    
            }
        }
    
        private List<String> getColumnNames(TableData table) throws SQLException {
            if (null != table && StringUtils.isNotBlank(table.getDatabase()) && StringUtils.isNotBlank(table.getTable())) {
                String key = table.getDatabase() + "." + table.getTable();
                if (!columnNamesMap.containsKey(key)) {
                    columnNamesMap.put(key, getColumnNames(key));
                }
                return columnNamesMap.get(key);
            }
            return null;
        }
    
        /**
         * 监听到的数据行转换为map
         *
         * @param rowsEventData
         * @param columnNames
         * @return
         */
        private List<Map<String, Object>> rowsToMap(Object rowsEventData, List<String> columnNames) {
            if (columnNames == null) {
                return null;
            }
            List<Map<String, Object>> datas = new ArrayList<>();
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("EEE MMM dd HH:mm:ss zzz yyyy", Locale.US);
            SimpleDateFormat simpleDateFormatGMT = new SimpleDateFormat("d MMM yyyy HH:mm:ss 'GMT'", Locale.US);
            try {
                Method method = rowsEventData.getClass().getMethod("getRows");
                Object value = method.invoke(rowsEventData);
                // 新增删除
                if (rowsEventData instanceof DeleteRowsEventData || rowsEventData instanceof WriteRowsEventData) {
                    for (Serializable[] row : (List<Serializable[]>) value) {
                        Map<String, Object> item = new HashMap<>();
                        for (int i = 0; i < columnNames.size(); i++) {
                            if (row[i] != null && row[i] instanceof Date) {
                                try {
                                    Date date = simpleDateFormat.parse(row[i].toString());
                                    item.put(columnNames.get(i), simpleDateFormatGMT.parse(((Date) row[i]).toGMTString()));
                                } catch (Exception e) {
                                    item.put(columnNames.get(i), row[i]);
                                }
                            } else {
                                item.put(columnNames.get(i), row[i]);
                            }
                        }
                        datas.add(item);
                    }
                } else {
                    for (Map.Entry<Serializable[], Serializable[]> row : (List<Map.Entry<Serializable[], Serializable[]>>) value) {
                        Map<String, Object> item = new HashMap<>();
                        for (int i = 0; i < columnNames.size(); i++) {
                            if (row.getValue()[i] != null && row.getValue()[i] instanceof Date) {
                                try {
                                    Date date = simpleDateFormat.parse(row.getValue()[i].toString());
                                    item.put(columnNames.get(i), simpleDateFormatGMT.parse(((Date) row.getValue()[i]).toGMTString()));
                                } catch (Exception e) {
                                    item.put(columnNames.get(i), row.getValue()[i]);
                                }
                            } else {
                                item.put(columnNames.get(i), row.getValue()[i]);
                            }
                        }
                        datas.add(item);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return datas;
        }
    
        /**
         * 取得该表字段名
         *
         * @param table
         * @return
         */
        public synchronized List<String> getColumnNames(String table) {
            try {
                List<String> columnNames = jdbcTemplate.query("SHOW FULL COLUMNS FROM " + table, new RowMapper<String>() {
                    @Override
                    public String mapRow(ResultSet resultSet, int i) throws SQLException {
                        return resultSet.getString("Field");
                    }
                });
                return columnNames;
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }
    
    }

#### 数据转换器 ####

将binlog数据转化为可用的java model对象

@Component
    @Slf4j
    public class BaseConverter implements Converter<BinlogData<List<Map<String, Object>>>> {
    
        @Autowired
        private ResourceLoader resourceLoader;
    
        @Autowired
        private JdbcTemplate jdbcTemplate;
    
        @Value("${binglog.convert.model.package}")
        private String packageName;
    
        @Autowired
        private CodeUtil codeUtil;
    
        @Autowired
        private DbProperties dbProperties;
    
        /**
         * 需要转成的model对象
         */
        private List<Class> classes;
    
        /**
         * 拼接sql常量
         */
        private static final String SQL_STATEMENT_SELECT = "SELECT ";
        private static final String SQL_STATEMENT_FROM = " FROM ";
        private static final String SQL_STATEMENT_WHERE = " WHERE ";
        private static final String SQL_STATEMENT_LIMIT = " LIMIT 1";
    
        @PostConstruct
        public void init() {
            classes = getClass(packageName, resourceLoader);
        }
    
        @Override
        public List convert(BinlogData<List<Map<String, Object>>> data) throws Exception {
            for (Class clazz : classes) {
                Table tableAnnotation = (Table) clazz.getAnnotation(Table.class);
                if (tableAnnotation != null && StringUtils.isNotBlank(tableAnnotation.name())) {
                    String dbName = dbProperties.getValue(tableAnnotation.catalog());//获取实际数据库名称
    
                    if (data.getDatabase().equals(dbName) && tableAnnotation.name().equals(data.getTable())) {
                        // 找到库 需要处理的表（data.getTable()）对应的model
                        return setBeanProperties(data.getData(), clazz);
                    }
                }
            }
            return null;
        }
    
        private List setBeanProperties(List<Map<String, Object>> list, Class clazz) {
            List target = new ArrayList();
            try {
                Field[] fields = clazz.getDeclaredFields();
                for (Object object : list) {
                    target.add(clazz.newInstance());
                }
                List<Field> fieldList = new ArrayList<>();
                for (Field field : fields) {
                    Column column = field.getAnnotation(Column.class);
                    if (column != null) {
                        // 该字段需要通过配置的库表查询
                        if ( StringUtils.isNotBlank(column.table())) {
                            // 需要查询该字段的值，本数据未包含该字段值,查询可能以来本对象其他字段值
                            // 先设置不需要查库的字段（查库字段可能以来这些字段得值）
                            // setFieldFromTable(field, target, column);
                            fieldList.add(field);
                        } else {
                            setFieldFromBinlog(field, target, list, column);
                        }
                    }
                }
                // cachedMap 用于缓存,多个字段值依赖同一条数据库数据
                Map<String, Map<String, Object>> cachedMap = new HashMap<>();
                for (Field field : fieldList) {
                    Column column = field.getAnnotation(Column.class);
                    if (column != null) {
                        // 该字段需要通过配置的库表查询
                        if ( StringUtils.isNotBlank(column.table()) && StringUtils.isNotBlank(column.whereCause())) {
                            // 需要查询该字段的值，本数据未包含该字段值
                            setFieldFromTable(field, cachedMap, list, target, column);
                        }
                    }
                }
            } catch (Exception e) {
                log.info("", e);
            }
            return target;
        }
    
        /**
         * TODO 类型转换
         *
         * @param field
         * @param target
         * @param list
         * @param column
         */
        private void setFieldFromBinlog(Field field, List target, List<Map<String, Object>> list, Column column) {
            if (StringUtils.isNotBlank(column.name())) {
                for (int i = 0; i < list.size(); i++) {
                    try {
                        Object value = list.get(i).get(column.name());
                        if (value != null) {
                            field.setAccessible(true);
                            log.debug("field:{},value:{}", field.getName(), column.name());
                            BinLogDict binLogDict = field.getAnnotation(BinLogDict.class);//判断是否需要翻译
                            if (binLogDict != null) {
                                Object companyCode = codeUtil.getCountryCode(binLogDict.codeType(), value, field.getType());
                                log.info("get companyCode fail value:{} companyCode:{}", value, companyCode);
                                field.set(target.get(i), companyCode);
                            } else {
                                field.set(target.get(i), value);
                            }
                            field.setAccessible(false);
                        }
                    } catch (Exception e) {
                        log.info("", e);
                    }
                }
            }
        }
    
        /**
         * TODO 数据库类型转换
         * 该字段从数据库查询设置
         * ${id} 参数取本对象的id 值
         * #{id} 参数取binlog 过来的id 值
         * <p>
         * <p>
         * // 本对象缓存（db.table.columnName.value）
         * // 本对象缓存（db.table.id.3）
         *
         * @param field
         * @param list
         * @param column
         */
        private void setFieldFromTable(Field field, Map<String, Map<String, Object>> cachedMap, List<Map<String, Object>> datas, List list, Column column) {
            try {
                field.setAccessible(true);
                String dbName = dbProperties.getValue(column.catalog());//获取实际数据库名称
    
                if (column.whereCause().contains("#")) {
                    // 需要依赖对象binlog字段值
                    int startLocation = column.whereCause().indexOf("#{") + 2;
                    int endLocation = column.whereCause().indexOf("}");
                    String refFieldName = column.whereCause().substring(startLocation, endLocation);
                    for (int i = 0; i < list.size(); i++) {
                        Object target = list.get(i);
                        StringBuilder sql = new StringBuilder();
                        sql.append(SQL_STATEMENT_SELECT).append("*").append(SQL_STATEMENT_FROM).append(dbName).append(".").append(column.table()).append(SQL_STATEMENT_WHERE);
                        Object refFieldValue = datas.get(i).get(refFieldName);
                        if (refFieldValue == null) {
                            continue;
                        }
                        // 本对象缓存（db.table.columnName.value）
                        // 本对象缓存（db.table.id.3）
                        String cacheKey = dbName + "." + column.table() + "." + refFieldName + "." + refFieldValue;
                        Map<String, Object> dataMap;
                        if (cachedMap.containsKey(cacheKey)) {
                            dataMap = cachedMap.get(cacheKey);
                            log.debug("命中缓存：cache_key:{}", cacheKey);
                        } else {
                            log.debug("cache_key:{}", cacheKey);
                            if (refFieldValue instanceof String) {
                                refFieldValue = "'" + refFieldValue + "'";
                            }
                            String whereCause = column.whereCause().replace("#{" + refFieldName + "}", String.valueOf(refFieldValue));
                            sql.append(whereCause).append(SQL_STATEMENT_LIMIT);
                            log.debug("查询字段：{}，sql：{}", field.getName(), sql);
                            dataMap = jdbcTemplate.queryForMap(sql.toString());
                            if (dataMap != null) {
                                cachedMap.put(cacheKey, dataMap);
                            }
                        }
                        if (dataMap != null) {
                            log.debug("field:{},value:{}", field.getName(), column.name());
                            Object value = dataMap.get(column.name());
    
                            BinLogDict binLogDict = field.getAnnotation(BinLogDict.class);//判断是否需要翻译
                            if (binLogDict != null) {
                                Object countryCode = codeUtil.getCountryCode(binLogDict.codeType(), value, field.getType());
                                log.info("get companyCode fail value:{} companyCode:{}", value, countryCode);
                                field.set(target, countryCode);
                            } else {
                                field.set(target, value);
                            }
                        }
                    }
                } else if (column.whereCause().contains("$")) {
                    // 需要依赖对象自身字段值
                    int startLocation = column.whereCause().indexOf("${") + 2;
                    int endLocation = column.whereCause().indexOf("}");
                    String refFieldName = column.whereCause().substring(startLocation, endLocation);
                    for (Object target : list) {
                        StringBuilder sql = new StringBuilder();
                        sql.append(SQL_STATEMENT_SELECT).append("*").append(SQL_STATEMENT_FROM).append(dbName).append(".").append(column.table()).append(SQL_STATEMENT_WHERE);
                        Field refField = target.getClass().getDeclaredField(refFieldName);
                        refField.setAccessible(true);
                        Object refFieldValue = refField.get(target);
                        if (refFieldValue == null) {
                            continue;
                        }
                        // 本对象缓存（db.table.columnName.value）
                        // 本对象缓存（db.table.id.3）
                        String cacheKey = dbName + "." + column.table() + "." + refFieldName + "." + refFieldValue;
                        Map<String, Object> dataMap;
                        if (cachedMap.containsKey(cacheKey)) {
                            dataMap = cachedMap.get(cacheKey);
                            log.debug("命中缓存：cache_key:{}", cacheKey);
                        } else {
                            if (refFieldValue instanceof String) {
                                refFieldValue = "'" + refFieldValue + "'";
                            }
                            String whereCause = column.whereCause().replace("${" + refFieldName + "}", String.valueOf(refFieldValue));
                            sql.append(whereCause).append(SQL_STATEMENT_LIMIT);
                            log.debug("查询字段：{}，sql：{}", field.getName(), sql);
                            dataMap = jdbcTemplate.queryForMap(sql.toString());
                            if (dataMap != null) {
                                cachedMap.put(cacheKey, dataMap);
                            }
                        }
                        if (dataMap != null) {
                            log.debug("field:{},value:{}", field.getName(), column.name());
                            Object value = dataMap.get(column.name());
    
                            BinLogDict binLogDict = field.getAnnotation(BinLogDict.class);//判断是否需要翻译
                            if (binLogDict != null) {
                                Object countryCode = codeUtil.getCountryCode(binLogDict.codeType(), value, field.getType());
                                log.info("get companyCode fail value:{} companyCode:{}", value, countryCode);
                                field.set(target, countryCode);
                            } else {
                                field.set(target, value);
                            }
                        }
                    }
                } else {
                    //  不需要依赖对象自身字段值
                    StringBuilder sql = new StringBuilder();
                    sql.append(SQL_STATEMENT_SELECT).append(column.name()).append(SQL_STATEMENT_FROM).append(dbName).append(".").append(column.table()).append(SQL_STATEMENT_WHERE);
                    sql.append(column.whereCause()).append(SQL_STATEMENT_LIMIT);
                    log.debug("查询字段：{}，sql：{}", field.getName(), sql);
                    Object object = jdbcTemplate.queryForObject(sql.toString(), field.getType());
                    for (Object target : list) {
                        log.debug("field:{},value:{}", field.getName(), column.name());
    
                        BinLogDict binLogDict = field.getAnnotation(BinLogDict.class);//判断是否需要翻译
                        if (binLogDict != null) {
                            Object countryCode = codeUtil.getCountryCode(binLogDict.codeType(), object, field.getType());
                            log.info("get companyCode fail value:{} companyCode:{}", object, countryCode);
                            field.set(target, countryCode);
                        } else {
                            field.set(target, object);
                        }
                    }
                }
            } catch (Exception e) {
                log.info("{}", e);
            } finally {
                field.setAccessible(false);
            }
    
        }
    
    
        /**
         * 根据报名取得改包下所有类
         *
         * @param packageName
         * @param resourceLoader
         * @return
         */
        private List<Class> getClass(String packageName, ResourceLoader resourceLoader) {
            List<Class> classList = new ArrayList<>();
            try {
                ResourcePatternResolver resolver = ResourcePatternUtils.getResourcePatternResolver(resourceLoader);
                MetadataReaderFactory metaReader = new CachingMetadataReaderFactory(resourceLoader);
                Resource[] resources = resolver.getResources("classpath*:" + packageName.replaceAll("\\.", "/") + "/*.class");
                for (Resource r : resources) {
                    MetadataReader reader = metaReader.getMetadataReader(r);
                    classList.add(Class.forName(reader.getClassMetadata().getClassName()));
                    log.debug("----->binlog会自动转化的model类型：{}", reader.getClassMetadata().getClassName());
                }
            } catch (Exception e) {
                log.info("", e);
            }
            return classList;
        }
    
    }

#### model 配置 ####

使用注解将binlog数据转化为model，每个字段都可以配置

@Data
    @Accessors(chain = true)
    @Table(catalog = "", name = "rcp_pc_teaching_receiver")
    public class TeachingData {
    
        /**
         * ${id} 参数取本对象的id 值
         * #{id} 参数取binlog 过来的id 值
         * 如果本对象多个字段依赖同一个sql和 条件查询结果的不同字段，不用担心做了缓存，只查一次
         * <p>
         * 缓存key（db.table.columnName.value
         * select $name from $catalog.$table where $whereCause
         */
        @Column(catalog = "cttms", table = "ca_t_human", name = "name", whereCause = "caid=#{human_caid}")
        private String name;
    
        @Column(catalog = "cttms_db_new_test", table = "ca_t_human", name = "traineeMajorCode", whereCause = "caid=#{human_caid}")
        private String traineeMajorCode;
    
        @Column(catalog = "cttms", table = "ca_t_human", name = "identificationNumber", whereCause = "caid=#{human_caid}")
        private String idCard;
    
        @Column(name = "department_caid")
        private String departmentId;
    
        @Column(catalog = "cttms", table = "rcp_pc_teaching", name = "start_time", whereCause = "id=#{teaching_id}")
        private Date scheduleStartTime;
    
        @Column(catalog = "cttms", table = "rcp_pc_teaching", name = "end_time", whereCause = "id=#{teaching_id}")
        private Date scheduleEndTime;
    
        private Integer stage;
    
        @Column(catalog = "cttms", table = "rcp_pc_teaching", name = "type", whereCause = "id=#{teaching_id}")
        private Integer activityType;
    
        @Column(catalog = "cttms", table = "rcp_pc_teaching", name = "teaching_level", whereCause = "id=#{teaching_id}")
        private Long activityLevel;
    
        @Column(catalog = "cttms", table = "rcp_pc_teaching", name = "detail", whereCause = "id=#{teaching_id}")
        private String activityContent;
    
        @Column(catalog = "cttms", table = "rcp_pc_teaching", name = "class_hour", whereCause = "id=#{teaching_id}")
        private Double activityDuration;
    
        @Column(catalog = "cttms", table = "rcp_pc_teaching", name = "speaker", whereCause = "id=#{teaching_id}")
        private String keynoteSpeaker;
    
        private String activityAttachments;
    
        @Column(catalog = "cttms", table = "ca_t_institute", name = "name", whereCause = "caid=#{institution_id}")
        private String baseName;
    
        @Column(name = "institution_id")
        private String baseCode;
    
        @Column(name = "id")
        private Long dataId;
    
        @Column(name = "human_caid")
        private String humanId;
    
        @Column(name = "is_del")
        private Integer isDel;
    
    }

#### 相关注解 ####

@Target({ElementType.TYPE})
    @Retention(RetentionPolicy.RUNTIME)
    public @interface Table {
    
        /**
         * 表名
         *
         * @return
         */
        String name() default "";
    
        /**
         * 库名
         *
         * @return
         */
        String catalog() default "";
    }
    
    
    
    /**
     * select $name from $catalog.$table where $whereCause
     */
    @Target({ElementType.METHOD, ElementType.FIELD})
    @Retention(RetentionPolicy.RUNTIME)
    public @interface Column {
    
        /**
         * 库名
         *
         * @return
         */
        String catalog() default "";
    
        /**
         * 表名
         *
         * @return
         */
        String table() default "";
    
        /**
         * 表属性名
         *
         * @return
         */
        String name() default "";
    
        /**
         * sql查询条件
         *
         * @return
         */
        String whereCause() default "";
    
    
    }