BinlogClientRunner.java 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231
  1. package com.persagy.binlog;
  2. import com.alibaba.fastjson.JSON;
  3. import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
  4. import com.baomidou.mybatisplus.core.metadata.IPage;
  5. import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
  6. import com.github.shyiko.mysql.binlog.BinaryLogClient;
  7. import com.github.shyiko.mysql.binlog.event.*;
  8. import com.persagy.ccontinue.entity.BinlogPosition;
  9. import com.persagy.business.HandlerData;
  10. import com.persagy.ccontinue.service.BinlogPositionService;
  11. import com.persagy.configuration.MyBatisPlusConfig;
  12. import com.persagy.entity.RwdObjectWd;
  13. import com.persagy.mapper.RwdObjectWdMapper;
  14. import com.persagy.utils.CacheUtil;
  15. import lombok.extern.slf4j.Slf4j;
  16. import org.springframework.beans.factory.annotation.Autowired;
  17. import org.springframework.beans.factory.annotation.Value;
  18. import org.springframework.boot.CommandLineRunner;
  19. import org.springframework.context.annotation.Configuration;
  20. import org.springframework.core.annotation.Order;
  21. import org.springframework.scheduling.annotation.Async;
  22. import java.io.Serializable;
  23. import java.security.NoSuchAlgorithmException;
  24. import java.security.SecureRandom;
  25. import java.util.Arrays;
  26. import java.util.HashMap;
  27. import java.util.List;
  28. import java.util.Map;
  29. @Slf4j
  30. @Configuration
  31. @Order(1000)
  32. public class BinlogClientRunner implements CommandLineRunner {
  33. @Autowired
  34. private HandlerData handlerData;
  35. @Autowired
  36. private BinlogPositionService binlogPositionService;
  37. @Autowired
  38. private RwdObjectWdMapper rwdObjectWdMapper;
  39. @Value("${binlog.host}")
  40. private String host;
  41. @Value("${binlog.port}")
  42. private int port;
  43. @Value("${binlog.user}")
  44. private String user;
  45. @Value("${binlog.password}")
  46. private String password;
  47. // binlog server_id
  48. @Value("${server.id}")
  49. private long serverId;
  50. // 指定监听的数据表
  51. @Value("${binlog.database.table}")
  52. private String database_table;
  53. // 指定监听的数据表
  54. @Value("${binlog.database.table-format}")
  55. private String tableFormat;
  56. @Async
  57. @Override
  58. public void run(String... args) throws Exception {
  59. // 获取监听数据表数组
  60. List<String> databaseList = Arrays.asList(database_table.split(","));
  61. //获取position的位置(创建client时,读取当前记录的postion),和查看是否是第一次启动,为空表示第一次启动
  62. BinlogPosition binlogPosition = binlogPositionService.getPosition();
  63. //先全量把数据同步一下再,增量同步数据
  64. //查询数据
  65. //表示需要同步到中台去的数据,先将监听表的infos字段信息查询出来
  66. if(binlogPosition==null){
  67. /**开始全量同步数据=======================================*/
  68. log.info("开始全量同步数据=======================================");
  69. for (String databaseTableName:databaseList
  70. ) {
  71. String tableName = databaseTableName.substring(databaseTableName.indexOf(".")+1);
  72. MyBatisPlusConfig.myTableName.set(tableName);
  73. //先查询有多少条数据
  74. QueryWrapper<RwdObjectWd> rwdObjectWdQueryWrapperCount = new QueryWrapper<>();
  75. rwdObjectWdQueryWrapperCount.in("class_code", CacheUtil.classCodeCacheList);
  76. Integer totalCount = rwdObjectWdMapper.selectCount(rwdObjectWdQueryWrapperCount);
  77. if(totalCount==null || totalCount==0){
  78. continue;
  79. }
  80. //分页查询
  81. //计算会分多少页
  82. int pageSize = 200;
  83. int pageTotal = totalCount/ pageSize;
  84. for (int i = 0; i <= pageTotal; i++) {
  85. //参数一是当前页,参数二是每页个数
  86. IPage<RwdObjectWd> rwdObjectWdPage = new Page<>(i, pageSize);
  87. QueryWrapper<RwdObjectWd> rwdObjectWdQueryWrapperOrder = new QueryWrapper<>();
  88. rwdObjectWdQueryWrapperOrder.orderByAsc("id");
  89. rwdObjectWdPage = rwdObjectWdMapper.selectPage(rwdObjectWdPage,rwdObjectWdQueryWrapperOrder);
  90. List<RwdObjectWd> rwdObjectWds = rwdObjectWdPage.getRecords();
  91. for (RwdObjectWd rwdObjectWd:rwdObjectWds
  92. ) {
  93. try {
  94. handlerData.handlerDataToWd(tableFormat,null,tableName,"insert",rwdObjectWd);
  95. } catch (Exception e) {
  96. log.error("全量同步数据失败:{}",e);
  97. }
  98. }
  99. log.info(tableName+"同步第"+i+"页,数量"+rwdObjectWds.size());
  100. }
  101. }
  102. if(binlogPosition==null){
  103. binlogPosition = new BinlogPosition();
  104. }
  105. binlogPosition.setServerId(serverId);
  106. binlogPositionService.saveOrUpdate(binlogPosition);
  107. }
  108. /**开始增量同步数据=======================================*/
  109. log.info("开始增量同步数据=======================================");
  110. HashMap<Long, String> tableMap = new HashMap<Long, String>();
  111. // 创建binlog监听客户端
  112. BinaryLogClient client = new BinaryLogClient(host, port, user, password);
  113. client.setServerId(serverId);
  114. if (binlogPosition != null &&
  115. binlogPosition.getBinlogName() != null &&
  116. binlogPosition.getPosition() != null) {
  117. client.setBinlogFilename(binlogPosition.getBinlogName());
  118. client.setBinlogPosition(binlogPosition.getPosition());
  119. }
  120. client.registerEventListener((event -> {
  121. // binlog事件
  122. EventData data = event.getData();
  123. if (data != null) {
  124. if (data instanceof TableMapEventData) {
  125. TableMapEventData tableMapEventData = (TableMapEventData) data;
  126. tableMap.put(tableMapEventData.getTableId(), tableMapEventData.getDatabase() + "." + tableMapEventData.getTable());
  127. }
  128. // update数据
  129. if (data instanceof UpdateRowsEventData) {
  130. UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
  131. String tableName = tableMap.get(updateRowsEventData.getTableId());
  132. if (tableName != null && databaseList.contains(tableName)) {
  133. String eventKey = tableName + ".update";
  134. log.info("监听数据库更新数据:{}",eventKey);
  135. for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
  136. List<Serializable> entries = Arrays.asList(row.getValue());
  137. try {
  138. handlerData.handlerDataToWd(tableFormat,entries,tableName,"update",null);
  139. } catch (Exception e) {
  140. log.error("同步数据失败:{}",e);
  141. }
  142. saveBinlogPosition(event);
  143. }
  144. }
  145. }
  146. // insert数据
  147. else if (data instanceof WriteRowsEventData) {
  148. WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
  149. String tableName = tableMap.get(writeRowsEventData.getTableId());
  150. if (tableName != null && databaseList.contains(tableName)) {
  151. String eventKey = tableName + ".insert";
  152. log.info("监听数据库插入数据:{}",eventKey);
  153. for (Serializable[] row : writeRowsEventData.getRows()) {
  154. List<Serializable> entries = Arrays.asList(row);
  155. try {
  156. handlerData.handlerDataToWd(tableFormat,entries,tableName,"insert",null);
  157. } catch (Exception e) {
  158. log.error("同步数据失败:{}",e);
  159. }
  160. saveBinlogPosition(event);
  161. }
  162. }
  163. }
  164. // delete数据
  165. else if (data instanceof DeleteRowsEventData) {
  166. DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
  167. String tableName = tableMap.get(deleteRowsEventData.getTableId());
  168. if (tableName != null && databaseList.contains(tableName)) {
  169. String eventKey = tableName + ".delete";
  170. for (Serializable[] row : deleteRowsEventData.getRows()) {
  171. System.out.println("delete======");
  172. }
  173. }
  174. }
  175. }
  176. }));
  177. client.connect();
  178. }
  179. private void saveBinlogPosition(Event event) {
  180. BinlogPosition binlogPositionSave = new BinlogPosition();
  181. //处理rotate事件,这里会替换调binlog fileName
  182. if (event.getHeader().getEventType().equals(EventType.ROTATE)) {
  183. RotateEventData rotateEventData = (RotateEventData) event.getData();
  184. binlogPositionSave.setBinlogName(rotateEventData.getBinlogFilename());
  185. binlogPositionSave.setPosition(rotateEventData.getBinlogPosition());
  186. binlogPositionSave.setServerId(event.getHeader().getServerId());
  187. } else {
  188. //统一处理事件对应的binlog position
  189. binlogPositionSave = binlogPositionService.getPosition();
  190. if(binlogPositionSave==null){
  191. binlogPositionSave = new BinlogPosition();
  192. }
  193. EventHeaderV4 eventHeaderV4 = (EventHeaderV4) event.getHeader();
  194. binlogPositionSave.setPosition(eventHeaderV4.getPosition());
  195. // binlogPositionSave.setBinlogName(event)
  196. binlogPositionSave.setServerId(event.getHeader().getServerId());
  197. }
  198. //将最新的配置保存到Redis中
  199. log.info("保存的数据{}", JSON.toJSONString(binlogPositionSave));
  200. binlogPositionService.saveOrUpdate(binlogPositionSave);
  201. }
  202. private static long getRandomServerId() {
  203. try {
  204. return SecureRandom.getInstanceStrong().nextLong();
  205. } catch (NoSuchAlgorithmException e) {
  206. // return RandomUtils.nextLong();
  207. }
  208. return 0L;
  209. }
  210. }