上一篇文章我们介绍了mysql binlog的基本用法，那么这票文章就带领大家优雅的使用mysql binlog在项目中使用

一、数据库配置

首先打开数据库的 binlog 支持，并且设置为 row 模式
查看是否支持binlog

  1. mysql> show variables like '%bin%';
  2. +--------------------------------------------+--------------------------------------+
  3. | Variable_name                              | Value                                |
  4. +--------------------------------------------+--------------------------------------+
  5. | bind_address                               | *                                    |
  6. | binlog_cache_size                          | 32768                                |
  7. | binlog_checksum                            | CRC32                                |
  8. | binlog_direct_non_transactional_updates    | OFF                                  |
  9. | binlog_error_action                        | ABORT_SERVER                         |
  10. | binlog_format                              | ROW                                  |
  11. | binlog_group_commit_sync_delay             | 0                                    |
  12. | binlog_group_commit_sync_no_delay_count    | 0                                    |
  13. | binlog_gtid_simple_recovery                | ON                                   |
  14. | binlog_max_flush_queue_time                | 0                                    |
  15. | binlog_order_commits                       | ON                                   |
  16. | binlog_row_image                           | FULL                                 |
  17. | binlog_rows_query_log_events               | OFF                                  |
  18. | binlog_stmt_cache_size                     | 32768                                |
  19. | binlog_transaction_dependency_history_size | 25000                                |
  20. | binlog_transaction_dependency_tracking     | COMMIT_ORDER                         |
  21. | innodb_api_enable_binlog                   | OFF                                  |
  22. | innodb_locks_unsafe_for_binlog             | OFF                                  |
  23. | log_bin                                    | ON                                   |
  24. | log_bin_basename                           | /data/mysql/mysql-5.7/data/bin       |
  25. | log_bin_index                              | /data/mysql/mysql-5.7/data/bin.index |
  26. | log_bin_trust_function_creators            | ON                                   |
  27. | log_bin_use_v1_row_events                  | OFF                                  |
  28. | log_statements_unsafe_for_binlog           | ON                                   |
  29. | max_binlog_cache_size                      | 18446744073709547520                 |
  30. | max_binlog_size                            | 1073741824                           |
  31. | max_binlog_stmt_cache_size                 | 18446744073709547520                 |
  32. | sql_log_bin                                | ON                                   |
  33. | sync_binlog                                | 1                                    |
  34. +--------------------------------------------+--------------------------------------+
  35. 29 rows in set (0.17 sec)

如果不支持,可通过在my.cnf中配置log-bin打开binlog支持，然后重启数据库再次查看是否支持

  1. log-bin=mysql-bin

相关变量

  1. -- Binlog 开关变量
  2. mysql> show variables like 'log_bin';
  3. +---------------+-------+
  4. | Variable_name | Value |
  5. +---------------+-------+
  6. | log_bin       | ON    |
  7. +---------------+-------+
  8. 1 row in set (0.30 sec)
  10. -- Binlog 日志的格式
  11. mysql> show variables like 'binlog_format';
  12. +---------------+-------+
  13. | Variable_name | Value |
  14. +---------------+-------+
  15. | binlog_format | ROW   |
  16. +---------------+-------+
  17. 1 row in set (0.00 sec)

二、创建项目

我们创建的是一个 springboot 的项目，不了解 springboot 的可以网上先了解下
然后我们引入 binlog 的 java 包

  1. <dependency>
  2.     <groupId>com.github.shyiko</groupId>
  3.     <artifactId>mysql-binlog-connector-java</artifactId>
  4.     <version>0.21.0</version>
  5. </dependency>

然后创建 监听处理器，这个处理器，主要处理项目中的应用注册到这个处理器上，并且监听注册到处理器中的数据库操作：

  1. import com.github.shyiko.mysql.binlog.BinaryLogClient;
  2. import com.github.shyiko.mysql.binlog.event.*;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.stereotype.Component;
  5. import org.springframework.util.StringUtils;
  7. import javax.annotation.Resource;
  8. import java.io.Serializable;
  9. import java.util.*;
  10. import java.util.stream.Collectors;
  12. /**
  13.  * @author clown
  14.  * @version 1.0.0
  15.  * @ClassName BinlogHandler
  16.  * @Description binlog 日志处理
  17.  */
  18. @Component
  19. @Slf4j
  20. public class BinlogHandler implements BinaryLogClient.EventListener {
  22.     private String dataBaseName;
  24.     private String tableName;
  26.     private final Map<String, BinLogListener> listenerMap = new HashMap<>();
  28.     @Resource
  29.     private BingLogTableHolder bingLogTableHolder;
  31.     private String getKey(String dataBaseName, String tableName) {
  32.         return dataBaseName + ":" + tableName;
  33.     }
  35.     public void register(String dataBaseName, String tableName, BinLogListener binLogListener) {
  36.         log.info("binlog handler register: {}-{}", dataBaseName, tableName);
  37.         this.listenerMap.put(getKey(dataBaseName, tableName), binLogListener);
  38.         bingLogTableHolder.initTable(dataBaseName, tableName);
  39.     }
  42.     @Override
  43.     public void onEvent(Event event) {
  44.         final EventType eventType = event.getHeader().getEventType();
  45.         if (eventType == EventType.TABLE_MAP) {
  46.             TableMapEventData data = event.getData();
  47.             this.tableName = data.getTable();
  48.             this.dataBaseName = data.getDatabase();
  49.             return;
  50.         }
  51.         if (eventType != EventType.UPDATE_ROWS
  52.                 && eventType != EventType.WRITE_ROWS
  53.                 && eventType != EventType.DELETE_ROWS) {
  54.             return;
  55.         }
  56.         if (StringUtils.isEmpty(dataBaseName) || StringUtils.isEmpty(tableName)) {
  57.             log.error("binlog handler error, dataBaseName or tableName is empty");
  58.             return;
  59.         }
  60.         // 找出对应表有兴趣的监听器
  61.         String key = getKey(this.dataBaseName, this.tableName);
  62.         BinLogListener listener = this.listenerMap.get(key);
  63.         if (null == listener) {
  64.             log.debug("skip {}", key);
  65.             return;
  66.         }
  67.         log.info("trigger event: {}", eventType.name());
  68.         try {
  69.             BingLogRowData rowData = this.build(event, listener);
  70.             rowData.setType(eventType);
  71.             rowData.setTableName(tableName);
  72.             rowData.setDataBaseName(dataBaseName);
  73.             listener.onEvent(rowData);
  74.         } catch (Exception e) {
  75.             log.error("binlog handler error ,error message:{}", e.getMessage(), e);
  76.         } finally {
  77.             this.dataBaseName = null;
  78.             this.tableName = null;
  79.         }
  80.     }
  82.     private List<Serializable[]> getAfterValues(EventData eventData) {
  83.         if (eventData instanceof WriteRowsEventData) {
  84.             return ((WriteRowsEventData) eventData).getRows();
  85.         }
  86.         if (eventData instanceof UpdateRowsEventData) {
  87.             return ((UpdateRowsEventData) eventData).getRows()
  88.                     .stream()
  89.                     .map(Map.Entry::getValue)
  90.                     .collect(Collectors.toList());
  91.         }
  92.         if (eventData instanceof DeleteRowsEventData) {
  93.             return ((DeleteRowsEventData) eventData).getRows();
  94.         }
  95.         return Collections.emptyList();
  96.     }
  98.     private List<Serializable[]> getBeforeValues(EventData eventData) {
  99.         if (eventData instanceof WriteRowsEventData) {
  100.             return null;
  101.         }
  102.         if (eventData instanceof UpdateRowsEventData) {
  103.             return ((UpdateRowsEventData) eventData).getRows()
  104.                     .stream()
  105.                     .map(Map.Entry::getKey)
  106.                     .collect(Collectors.toList());
  107.         }
  108.         if (eventData instanceof DeleteRowsEventData) {
  109.             return null;
  110.         }
  111.         return Collections.emptyList();
  112.     }
  114.     private BingLogRowData build(Event event, BinLogListener listener) {
  115.         final EventData data = event.getData();
  116.         BingLogRowData rowData = new BingLogRowData();
  117.         if (data instanceof WriteRowsEventData) {
  118.             WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
  119.             listener.onWrite(writeRowsEventData);
  120.         } else if (data instanceof UpdateRowsEventData) {
  121.             UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
  122.             listener.onUpdate(updateRowsEventData);
  123.         } else if (data instanceof DeleteRowsEventData) {
  124.             DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
  125.             listener.onDelete(deleteRowsEventData);
  126.         }
  127.         final Map<Integer, String> table = bingLogTableHolder.getTable(this.dataBaseName, this.tableName);
  128.         List<Map<String, Object>> afterMapList = new ArrayList<>();
  129.         List<Map<String, Object>> beforeMapList = new ArrayList<>();
  130.         final List<Serializable[]> beforeValues = getBeforeValues(data);
  131.         final List<Serializable[]> afterValues = getAfterValues(data);
  132.         if (afterValues != null) {
  133.             for (int i = 0; i < afterValues.size(); i++) {
  134.                 Map<String, Object> afterMap = new HashMap<>();
  135.                 Map<String, Object> beforeMap = new HashMap<>();
  136.                 final Serializable[] after = afterValues.get(i);
  137.                 Serializable[] before = null;
  138.                 if (beforeValues != null) {
  139.                     before = beforeValues.get(i);
  140.                 }
  141.                 for (int ix = 0; ix < after.length; ix++) {
  142.                     String colName = table.get(ix + 1);
  143.                     if (colName == null) {
  144.                         continue;
  145.                     }
  146.                     Object afterValue = after[ix];
  147.                     afterMap.put(colName, afterValue);
  148.                     if ("id".equals(colName)){
  149.                         rowData.setDataId(afterValue);
  150.                     }
  151.                     if (before != null) {
  152.                         Object beforeValue = before[ix];
  153.                         beforeMap.put(colName, beforeValue);
  154.                         beforeMapList.add(beforeMap);
  155.                         rowData.setBefore(beforeMapList);
  156.                     }
  157.                     afterMapList.add(afterMap);
  158.                     rowData.setAfter(afterMapList);
  159.                 }
  160.             }
  161.         }
  162.         return rowData;
  163.     }
  164. }

上面是 binlog 的处理器，其中 register 方法是别的监听器注册的方法，当别的监听器注册到这个处理器上后，这个处理器就可以处理那些对应数据库数据更新监听，并且初始化其监听的数据库对应的表的字段数据

onEvent 方法 则是实现 BinaryLogClient.EventListener 的方法，当 BinaryLogClient 中监听的事件方法，此方法实现了获取监听到的数据，并将数据处理后分发给对应的监听器，让对应监听器处理其对应的数据变动

下面是对应的 binlog 数据库表字段监听的初始化类

  1. import lombok.Getter;
  2. import lombok.Setter;
  3. import org.springframework.jdbc.core.JdbcTemplate;
  4. import org.springframework.jdbc.core.RowMapper;
  5. import org.springframework.stereotype.Component;
  7. import javax.annotation.Resource;
  8. import java.sql.ResultSet;
  9. import java.sql.SQLException;
  10. import java.util.HashMap;
  11. import java.util.List;
  12. import java.util.Map;
  13. import java.util.stream.Collectors;
  15. /**
  16.  * @author clown
  17.  * @version 1.0.0
  18.  * @ClassName BingLogTableHolder
  19.  * @Description binlog 处理的表字段
  20.  */
  21. @Component
  22. public class BingLogTableHolder {
  24.     @Resource
  25.     private JdbcTemplate jdbcTemplate;
  27.     private static final String INIT_SQL = "select table_schema, table_name, column_name, " +
  28.             "ordinal_position from information_schema.columns where table_schema = ? and table_name = ?";
  30.     private static final Map<String, Map<Integer, String>> TABLE_COLUMN_MAP = new HashMap<>();
  32.     public void initTable(String databaseName, String tableName) {
  33.         final List<TableColumn> query = jdbcTemplate.query(INIT_SQL, new Object[]{databaseName, tableName}, (rs, index) -> new TableColumn().mapRow(rs, index));
  34.         final Map<Integer, String> collect = query.stream()
  35.                 .collect(Collectors.toMap(TableColumn::getPosition, TableColumn::getColumnName));
  36.         TABLE_COLUMN_MAP.put(databaseName + ":" + tableName, collect);
  37.     }
  39.     public Map<Integer, String> getTable(String databaseName, String tableName) {
  40.         return TABLE_COLUMN_MAP.get(databaseName + ":" + tableName);
  41.     }
  43.     @Getter
  44.     @Setter
  45.     public static class TableColumn implements RowMapper<TableColumn> {
  46.         private String columnName;
  47.         private Integer position;
  48.         //private String tableName;
  49.         //private String databaseName;
  51.         @Override
  52.         public TableColumn mapRow(ResultSet rs, int i) throws SQLException {
  53. //            this.databaseName = rs.getString("table_schema");
  54. //            this.tableName = rs.getString("table_name");
  55.             this.columnName = rs.getString("column_name");
  56.             this.position = rs.getInt("ordinal_position");
  57.             return this;
  58.         }
  59.     }
  60. }

上面的类比较简单，核心就两个方法，initTable 就是根据传入的数据库名称和表名称，将数据库中表的字段名和其所处的数据库中存储postion获取出来，因为binog 只能获取到数据在表中所处的第n个字段，所以需要这样处理

getTable就是根据传入的数据库名称和表名称获取当前表中字段的下标
接着就是监听器了

首先是监听器的接口

  1. import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
  2. import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData;
  3. import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;
  5. /**
  6.  * @author clown
  7.  * @version 1.0.0
  8.  * @ClassName BinLogListener
  9.  * @Description bin
  10.  */
  11. public interface BinLogListener {
  13.     /**
  14.      * 将监听器注册到处理器中
  15.      */
  16.     void register(BinlogHandler binlogHandler);
  18.     /**
  19.      * 当触发这些事件的时候
  20.      *
  21.      * @param rowData
  22.      */
  23.     void onEvent(BingLogRowData rowData) throws Exception;
  25.     default void onUpdate(UpdateRowsEventData eventData) {
  26.         System.out.println("UpdateRowsEventData:" +eventData);
  27.     }
  29.     default void onDelete(DeleteRowsEventData eventData) {
  30.         System.out.println("DeleteRowsEventData:" +eventData);
  31.     }
  33.     default void onWrite(WriteRowsEventData eventData) {
  34.         System.out.println("WriteRowsEventData:" +eventData);
  35.     }

上面的 register 方法就是将此监听器注册到handler中的工具
onEvent方法 则是根据 handler 中的事件和监听类型分发处理的方法
其他的几个 默认方法是触发对应的事件后的方法，此处不再多加说明
然后是监听器的实现类

  1. import com.share51.cms.binlog.BinLogListener;
  2. import com.share51.cms.binlog.BingLogRowData;
  3. import com.share51.cms.binlog.BinlogHandler;
  4. import com.share51.cms.entity.Link;
  5. import org.springframework.stereotype.Component;
  7. /**
  8.  * @author clown
  9.  * @version 1.0.0
  10.  * @ClassName LinkListener
  11.  * @Description 友情管理监听器
  12.  */
  13. @Component("linkListener")
  14. public class LinkListener implements BinLogListener {
  16.     private static final String DATA_BASE_NAME = "cms_study";
  17.     private static final String TABLE_NAME = "t_link";
  19.     @Override
  20.     public void register(BinlogHandler binlogHandler) {
  21.         binlogHandler.register(DATA_BASE_NAME, TABLE_NAME, this);
  22.     }
  24.     @Override
  25.     public void onEvent(BingLogRowData eventData) throws Exception {
  26.         System.out.println(this.getClass().getName() + ":" + eventData);
  27.         final Link beforeData = eventData.getBeforeData(Link.class);
  28.         System.out.println(beforeData.getId());
  29.     }
  31. }

此处不再多加描述，就是将 DATA_BASE_NAME 库和 TABLE_NAME 表注册到监听器中，让后处理其方法
接着是 handler 处理的数据变化后对象

  1. import com.github.shyiko.mysql.binlog.event.EventType;
  2. import com.share51.cms.entity.BaseEntity;
  3. import lombok.Getter;
  4. import lombok.Setter;
  5. import lombok.ToString;
  7. import java.lang.reflect.Field;
  8. import java.util.*;
  9. import java.util.stream.Collectors;
  10. import java.util.stream.Stream;
  12. /**
  13.  * @author clown
  14.  * @version 1.0.0
  15.  * @ClassName BingLogRowData
  16.  * @Description binlog数据
  17.  */
  18. @Getter
  19. @Setter
  20. @ToString
  21. public class BingLogRowData {
  23.     /**
  24.      * 数据库的名称
  25.      */
  26.     private String dataBaseName;
  28.     /**
  29.      * 监听到变化表的名称
  30.      */
  31.     private String tableName;
  33.     /**
  34.      * 修改数据库的主键 id
  35.      */
  36.     private Object dataId;
  38.     /**
  39.      * 监听的 binglog 类型
  40.      */
  41.     private EventType type;
  43.     /**
  44.      * 修改前的数据
  45.      */
  46.     private List<Map<String, Object>> after;
  48.     /**
  49.      * 修改后的数据
  50.      */
  51.     private List<Map<String, Object>> before;
  53.     public <T extends BaseEntity> T getAfterData(Class<T> clazz) throws Exception {
  54.         return getData(after, clazz);
  55.     }
  57.     public <T extends BaseEntity> T getBeforeData(Class<T> clazz) throws Exception {
  58.         return getData(before, clazz);
  59.     }
  61.     private <T extends BaseEntity> T getData(List<Map<String, Object>> data, Class<T> clazz) throws IllegalAccessException, InstantiationException {
  62.         final T t = clazz.newInstance();
  63.         if (data != null) {
  64.             data.forEach(map -> {
  65.                 for (String key : map.keySet()) {
  66.                     final String fieldName = getFieldName(key);
  67.                     Field declaredField = null;
  68.                     try {
  69.                         declaredField = clazz.getDeclaredField(fieldName);
  70.                     } catch (NoSuchFieldException e) {
  71.                         try {
  72.                             declaredField = clazz.getSuperclass().getDeclaredField(fieldName);
  73.                         } catch (NoSuchFieldException e1) {
  74.                             e1.printStackTrace();
  75.                         }
  76.                     }
  77.                     try {
  78.                         if (declaredField != null) {
  79.                             declaredField.setAccessible(true);
  80.                             declaredField.set(t, map.get(key));
  81.                             declaredField.setAccessible(false);
  82.                         }
  83.                     } catch (IllegalAccessException e) {
  84.                         e.printStackTrace();
  85.                     }
  86.                 }
  87.             });
  88.         }
  89.         return t;
  90.     }
  92.     private String getFieldName(String columnName) {
  93.         final String[] split = columnName.split("_");
  94.         final String collect = Stream.of(split)
  95.                 .map(text ->
  96.                         text.substring(0, 1).toUpperCase() + text.substring(1, text.length())
  97.                 ).collect(Collectors.joining());
  98.         return collect.substring(0, 1).toLowerCase() + collect.substring(1, collect.length());
  99.     }
  101. }

上面的代码比较简单，此处就不多加描述，只是数据的处理交换层，相当于vo对象
然后是项目binlog的启动和关闭类

  1. import com.github.shyiko.mysql.binlog.BinaryLogClient;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.stereotype.Component;
  5. import javax.annotation.PostConstruct;
  6. import javax.annotation.PreDestroy;
  7. import javax.annotation.Resource;
  8. import java.io.IOException;
  9. import java.util.Map;
  10. import java.util.Set;
  12. /**
  13.  * @author clown
  14.  * @version 1.0.0
  15.  * @ClassName BinlogClient
  16.  * @Description binlog 客户端
  17.  */
  18. @Component
  19. public class BinlogClient {
  21.     @Resource
  22.     private BinaryLogClient binaryLogClient;
  24.     @Autowired
  25.     private Map<String, BinLogListener> binLogListenerMap;
  27.     @Resource
  28.     private BinlogHandler binlogHandler;
  30.     @PostConstruct
  31.     public void start() {
  32.         // 注册监听器
  33.         final Set<Map.Entry<String, BinLogListener>> entries = binLogListenerMap.entrySet();
  34.         for (Map.Entry<String, BinLogListener> entity : entries) {
  35.             final BinLogListener listener = entity.getValue();
  36.             listener.register(binlogHandler);
  37.         }
  38.         new Thread(() -> {
  39.             binaryLogClient.registerEventListener(binlogHandler);
  40.             try {
  41.                 binaryLogClient.connect();
  42.             } catch (IOException e) {
  43.                 e.printStackTrace();
  44.             }
  45.         }).start();
  46.     }
  48.     @PreDestroy
  49.     public void destroy() throws IOException {
  50.         if (binaryLogClient != null) {
  51.             binaryLogClient.disconnect();
  52.         }
  53.     }

此处也是比较简单的方法，只不过，此处用到了 spring 一个比较容易被忽视的功能，依赖查询，binLogListenerMap 中实际存储的是 实现 binLogListener 接口的所有bean，此处这么使用的原因是为了快速找到项目中的所有监听器，并且将所有的监听器注册到handler中。

然后在线程中启动这个对象的原因是，binlog 会造成线程的堵塞，若不在子线程中启动，就会一直堵塞项目启动。

最后就是客户端的配置了

  1. import com.github.shyiko.mysql.binlog.BinaryLogClient;
  2. import com.github.shyiko.mysql.binlog.event.deserialization.*;
  3. import lombok.Getter;
  4. import lombok.Setter;
  5. import org.springframework.boot.context.properties.ConfigurationProperties;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.context.annotation.Configuration;
  9. /**
  10.  * @author clown
  11.  * @version 1.0.0
  12.  * @ClassName BinlogConfig
  13.  * @Description mysql binlog 配置类
  14.  */
  15. @Getter
  16. @Setter
  17. @Configuration
  18. @ConfigurationProperties(prefix = "share51.binlog")
  19. public class BinlogConfig {
  21.     private String host;
  23.     private Integer port = 3306;
  25.     private String username;
  27.     private String password;
  29.     @Bean
  30.     public BinaryLogClient binaryLogClient() {
  31.         BinaryLogClient client = new BinaryLogClient(host, port, username, password);
  32.         EventDeserializer eventDeserializer = new EventDeserializer();
  33.         eventDeserializer.setCompatibilityMode(
  34.                 EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
  35.                 EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
  36.         );
  37.         return client;
  38.     }

以上就是 binlog 的主要代码了，想要处理数据缓存的话，可以在每个监听器的 onEvent 中处理对应的缓存更新业务。