BinlogClientRunnerCopy.java 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123
  1. //package com.persagy.cn.binlog;
  2. //
  3. //import com.alibaba.fastjson.JSON;
  4. //import com.alibaba.fastjson.JSONObject;
  5. //import com.github.shyiko.mysql.binlog.BinaryLogClient;
  6. //import com.github.shyiko.mysql.binlog.event.*;
  7. //import com.persagy.cn.business.HandlerData;
  8. //import lombok.extern.slf4j.Slf4j;
  9. //import org.springframework.beans.factory.annotation.Autowired;
  10. //import org.springframework.beans.factory.annotation.Value;
  11. //import org.springframework.boot.CommandLineRunner;
  12. //import org.springframework.context.annotation.Configuration;
  13. //import org.springframework.core.annotation.Order;
  14. //import org.springframework.scheduling.annotation.Async;
  15. //
  16. //import java.io.Serializable;
  17. //import java.util.Arrays;
  18. //import java.util.HashMap;
  19. //import java.util.List;
  20. //import java.util.Map;
  21. //
  22. ///**
  23. // * @Author : weiyizhong
  24. // * @Description: binlog监听
  25. // * @Date : 2021/9/15 10:37
  26. // * @Modified By :
  27. // */
  28. //@Slf4j
  29. //@Configuration
  30. //@Order(1000)
  31. //public class BinlogClientRunner implements CommandLineRunner {
  32. //
  33. // @Autowired
  34. // private HandlerData handlerData;
  35. //
  36. // @Value("${binlog.host}")
  37. // private String host;
  38. //
  39. // @Value("${binlog.port}")
  40. // private int port;
  41. //
  42. // @Value("${binlog.user}")
  43. // private String user;
  44. //
  45. // @Value("${binlog.password}")
  46. // private String password;
  47. //
  48. // // binlog server_id
  49. // @Value("${server.id}")
  50. // private long serverId;
  51. //
  52. // // 指定监听的数据表
  53. // @Value("${binlog.database.table}")
  54. // private String database_table;
  55. //
  56. // // 指定监听的数据表
  57. // @Value("${binlog.database.table-format}")
  58. // private String tableFormat;
  59. //
  60. //
  61. // @Async
  62. // @Override
  63. // public void run(String... args) throws Exception {
  64. // // 获取监听数据表数组
  65. // List<String> databaseList = Arrays.asList(database_table.split(","));
  66. // HashMap<Long, String> tableMap = new HashMap<Long, String>();
  67. // // 创建binlog监听客户端
  68. // BinaryLogClient client = new BinaryLogClient(host, port, user, password);
  69. // client.setServerId(serverId);
  70. //// client.setBinlogFilename("mysql-bin.000053");
  71. //// client.setBinlogPosition(0);
  72. // client.registerEventListener((event -> {
  73. // // binlog事件
  74. // EventData data = event.getData();
  75. // if (data != null) {
  76. // if (data instanceof TableMapEventData) {
  77. // TableMapEventData tableMapEventData = (TableMapEventData) data;
  78. // tableMap.put(tableMapEventData.getTableId(), tableMapEventData.getDatabase() + "." + tableMapEventData.getTable());
  79. // }
  80. // // update数据
  81. // if (data instanceof UpdateRowsEventData) {
  82. // UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
  83. // String tableName = tableMap.get(updateRowsEventData.getTableId());
  84. // if (tableName != null && databaseList.contains(tableName)) {
  85. // String eventKey = tableName + ".update";
  86. // log.info("监听数据库更新数据:{}",eventKey);
  87. // for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
  88. // List<Serializable> entries = Arrays.asList(row.getValue());
  89. // handlerData.handlerDataToWd(tableFormat,entries,tableName,"update");
  90. // }
  91. // }
  92. // }
  93. // // insert数据
  94. // else if (data instanceof WriteRowsEventData) {
  95. // WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
  96. // String tableName = tableMap.get(writeRowsEventData.getTableId());
  97. // if (tableName != null && databaseList.contains(tableName)) {
  98. // String eventKey = tableName + ".insert";
  99. // log.info("监听数据库插入数据:{}",eventKey);
  100. // for (Serializable[] row : writeRowsEventData.getRows()) {
  101. // List<Serializable> entries = Arrays.asList(row);
  102. // handlerData.handlerDataToWd(tableFormat,entries,tableName,"insert");
  103. // }
  104. // }
  105. // }
  106. // // delete数据
  107. // else if (data instanceof DeleteRowsEventData) {
  108. // DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
  109. // String tableName = tableMap.get(deleteRowsEventData.getTableId());
  110. // if (tableName != null && databaseList.contains(tableName)) {
  111. // String eventKey = tableName + ".delete";
  112. // for (Serializable[] row : deleteRowsEventData.getRows()) {
  113. // System.out.println("delete======");
  114. // }
  115. // }
  116. // }
  117. // }
  118. // }));
  119. // client.connect();
  120. // client.disconnect();
  121. // }
  122. //
  123. //}