123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123 |
- //package com.persagy.cn.binlog;
- //
- //import com.alibaba.fastjson.JSON;
- //import com.alibaba.fastjson.JSONObject;
- //import com.github.shyiko.mysql.binlog.BinaryLogClient;
- //import com.github.shyiko.mysql.binlog.event.*;
- //import com.persagy.cn.business.HandlerData;
- //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.util.Arrays;
- //import java.util.HashMap;
- //import java.util.List;
- //import java.util.Map;
- //
- ///**
- // * @Author : weiyizhong
- // * @Description: binlog监听
- // * @Date : 2021/9/15 10:37
- // * @Modified By :
- // */
- //@Slf4j
- //@Configuration
- //@Order(1000)
- //public class BinlogClientRunner implements CommandLineRunner {
- //
- // @Autowired
- // private HandlerData handlerData;
- //
- // @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(","));
- // HashMap<Long, String> tableMap = new HashMap<Long, String>();
- // // 创建binlog监听客户端
- // BinaryLogClient client = new BinaryLogClient(host, port, user, password);
- // client.setServerId(serverId);
- //// client.setBinlogFilename("mysql-bin.000053");
- //// client.setBinlogPosition(0);
- // 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());
- // handlerData.handlerDataToWd(tableFormat,entries,tableName,"update");
- // }
- // }
- // }
- // // 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);
- // handlerData.handlerDataToWd(tableFormat,entries,tableName,"insert");
- // }
- // }
- // }
- // // 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();
- // client.disconnect();
- // }
- //
- //}
|