package com.persagy.binlog; import com.alibaba.fastjson.JSON; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.core.metadata.IPage; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.github.shyiko.mysql.binlog.BinaryLogClient; import com.github.shyiko.mysql.binlog.event.*; import com.persagy.ccontinue.entity.BinlogPosition; import com.persagy.business.HandlerData; import com.persagy.ccontinue.service.BinlogPositionService; import com.persagy.configuration.MyBatisPlusConfig; import com.persagy.entity.RwdObjectWd; import com.persagy.mapper.RwdObjectWdMapper; import com.persagy.utils.CacheUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.CommandLineRunner; import org.springframework.context.annotation.Configuration; import org.springframework.core.annotation.Order; import org.springframework.scheduling.annotation.Async; import java.io.Serializable; import java.security.NoSuchAlgorithmException; import java.security.SecureRandom; import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @Slf4j @Configuration @Order(1000) public class BinlogClientRunner implements CommandLineRunner { @Autowired private HandlerData handlerData; @Autowired private BinlogPositionService binlogPositionService; @Autowired private RwdObjectWdMapper rwdObjectWdMapper; @Value("${binlog.host}") private String host; @Value("${binlog.port}") private int port; @Value("${binlog.user}") private String user; @Value("${binlog.password}") private String password; // binlog server_id @Value("${server.id}") private long serverId; // 指定监听的数据表 @Value("${binlog.database.table}") private String database_table; // 指定监听的数据表 @Value("${binlog.database.table-format}") private String tableFormat; @Async @Override public void run(String... args) throws Exception { // 获取监听数据表数组 List databaseList = Arrays.asList(database_table.split(",")); //获取position的位置(创建client时,读取当前记录的postion),和查看是否是第一次启动,为空表示第一次启动 BinlogPosition binlogPosition = binlogPositionService.getPosition(); //先全量把数据同步一下再,增量同步数据 //查询数据 //表示需要同步到中台去的数据,先将监听表的infos字段信息查询出来 if(binlogPosition==null){ /**开始全量同步数据=======================================*/ log.info("开始全量同步数据======================================="); for (String databaseTableName:databaseList ) { String tableName = databaseTableName.substring(databaseTableName.indexOf(".")+1); MyBatisPlusConfig.myTableName.set(tableName); //先查询有多少条数据 QueryWrapper rwdObjectWdQueryWrapperCount = new QueryWrapper<>(); rwdObjectWdQueryWrapperCount.in("class_code", CacheUtil.classCodeCacheList); Integer totalCount = rwdObjectWdMapper.selectCount(rwdObjectWdQueryWrapperCount); if(totalCount==null || totalCount==0){ continue; } //分页查询 //计算会分多少页 int pageSize = 200; int pageTotal = totalCount/ pageSize; for (int i = 0; i <= pageTotal; i++) { //参数一是当前页,参数二是每页个数 IPage rwdObjectWdPage = new Page<>(i, pageSize); QueryWrapper rwdObjectWdQueryWrapperOrder = new QueryWrapper<>(); rwdObjectWdQueryWrapperOrder.orderByAsc("id"); rwdObjectWdPage = rwdObjectWdMapper.selectPage(rwdObjectWdPage,rwdObjectWdQueryWrapperOrder); List rwdObjectWds = rwdObjectWdPage.getRecords(); for (RwdObjectWd rwdObjectWd:rwdObjectWds ) { try { handlerData.handlerDataToWd(tableFormat,null,tableName,"insert",rwdObjectWd); } catch (Exception e) { log.error("全量同步数据失败:{}",e); } } log.info(tableName+"同步第"+i+"页,数量"+rwdObjectWds.size()); } } if(binlogPosition==null){ binlogPosition = new BinlogPosition(); } binlogPosition.setServerId(serverId); binlogPositionService.saveOrUpdate(binlogPosition); } /**开始增量同步数据=======================================*/ log.info("开始增量同步数据======================================="); HashMap tableMap = new HashMap(); // 创建binlog监听客户端 BinaryLogClient client = new BinaryLogClient(host, port, user, password); client.setServerId(serverId); if (binlogPosition != null && binlogPosition.getBinlogName() != null && binlogPosition.getPosition() != null) { client.setBinlogFilename(binlogPosition.getBinlogName()); client.setBinlogPosition(binlogPosition.getPosition()); } client.registerEventListener((event -> { // binlog事件 EventData data = event.getData(); if (data != null) { if (data instanceof TableMapEventData) { TableMapEventData tableMapEventData = (TableMapEventData) data; tableMap.put(tableMapEventData.getTableId(), tableMapEventData.getDatabase() + "." + tableMapEventData.getTable()); } // update数据 if (data instanceof UpdateRowsEventData) { UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data; String tableName = tableMap.get(updateRowsEventData.getTableId()); if (tableName != null && databaseList.contains(tableName)) { String eventKey = tableName + ".update"; log.info("监听数据库更新数据:{}",eventKey); for (Map.Entry row : updateRowsEventData.getRows()) { List entries = Arrays.asList(row.getValue()); try { handlerData.handlerDataToWd(tableFormat,entries,tableName,"update",null); } catch (Exception e) { log.error("同步数据失败:{}",e); } saveBinlogPosition(event); } } } // insert数据 else if (data instanceof WriteRowsEventData) { WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data; String tableName = tableMap.get(writeRowsEventData.getTableId()); if (tableName != null && databaseList.contains(tableName)) { String eventKey = tableName + ".insert"; log.info("监听数据库插入数据:{}",eventKey); for (Serializable[] row : writeRowsEventData.getRows()) { List entries = Arrays.asList(row); try { handlerData.handlerDataToWd(tableFormat,entries,tableName,"insert",null); } catch (Exception e) { log.error("同步数据失败:{}",e); } saveBinlogPosition(event); } } } // delete数据 else if (data instanceof DeleteRowsEventData) { DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data; String tableName = tableMap.get(deleteRowsEventData.getTableId()); if (tableName != null && databaseList.contains(tableName)) { String eventKey = tableName + ".delete"; for (Serializable[] row : deleteRowsEventData.getRows()) { System.out.println("delete======"); } } } } })); client.connect(); } private void saveBinlogPosition(Event event) { BinlogPosition binlogPositionSave = new BinlogPosition(); //处理rotate事件,这里会替换调binlog fileName if (event.getHeader().getEventType().equals(EventType.ROTATE)) { RotateEventData rotateEventData = (RotateEventData) event.getData(); binlogPositionSave.setBinlogName(rotateEventData.getBinlogFilename()); binlogPositionSave.setPosition(rotateEventData.getBinlogPosition()); binlogPositionSave.setServerId(event.getHeader().getServerId()); } else { //统一处理事件对应的binlog position binlogPositionSave = binlogPositionService.getPosition(); if(binlogPositionSave==null){ binlogPositionSave = new BinlogPosition(); } EventHeaderV4 eventHeaderV4 = (EventHeaderV4) event.getHeader(); binlogPositionSave.setPosition(eventHeaderV4.getPosition()); // binlogPositionSave.setBinlogName(event) binlogPositionSave.setServerId(event.getHeader().getServerId()); } //将最新的配置保存到Redis中 log.info("保存的数据{}", JSON.toJSONString(binlogPositionSave)); binlogPositionService.saveOrUpdate(binlogPositionSave); } private static long getRandomServerId() { try { return SecureRandom.getInstanceStrong().nextLong(); } catch (NoSuchAlgorithmException e) { // return RandomUtils.nextLong(); } return 0L; } }