123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231 |
- 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<String> 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<RwdObjectWd> 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<RwdObjectWd> rwdObjectWdPage = new Page<>(i, pageSize);
- QueryWrapper<RwdObjectWd> rwdObjectWdQueryWrapperOrder = new QueryWrapper<>();
- rwdObjectWdQueryWrapperOrder.orderByAsc("id");
- rwdObjectWdPage = rwdObjectWdMapper.selectPage(rwdObjectWdPage,rwdObjectWdQueryWrapperOrder);
- List<RwdObjectWd> 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<Long, String> tableMap = new HashMap<Long, String>();
- // 创建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<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
- List<Serializable> 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<Serializable> 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;
- }
- }
|