基于canal的实时数据同步

傷城~ 2022-04-12 05:01 376阅读 0赞

适用场景

使用canal做数据备份而不用mysql自带的主从备份的场景主要为：

  1. 跨数据库的数据备份，例如mysql => oracle
  2. 数据异构，即对同一份数据做不同的分库分表查询。例如卖家和买家各自分库索引

maven

  1. <dependency>
  2.     <groupId>com.alibaba.otter</groupId>
  3.     <artifactId>canal.client</artifactId>
  4.     <version>1.1.2</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>com.alibaba.otter</groupId>
  8.     <artifactId>canal.protocol</artifactId>
  9.     <version>1.1.2</version>
  10. </dependency>

java

  1. import java.net.InetSocketAddress;
  2. import java.util.List;
  4. import com.alibaba.otter.canal.client.CanalConnector;
  5. import com.alibaba.otter.canal.client.CanalConnectors;
  6. import com.alibaba.otter.canal.protocol.CanalEntry;
  7. import com.alibaba.otter.canal.protocol.CanalEntry.Column;
  8. import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
  9. import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
  10. import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
  11. import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
  12. import com.alibaba.otter.canal.protocol.Message;
  13. import org.apache.commons.lang.StringUtils;
  15. public class SimpleCanalClient { 
  17.     public static void main(String[] args) throws Exception { 
  18.         String destination = "example";
  19.         CanalConnector connector = CanalConnectors.newSingleConnector(
  20.                 new InetSocketAddress("127.0.0.1", 11111), destination, "", "");
  22.         connector.connect();
  23.         connector.subscribe(".*\\..*");
  24.         connector.rollback();
  26.         int batchSize = 5 * 1024;
  28.         while (true) { 
  29.             Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
  30.             long batchId = message.getId();
  31.             int size = message.getEntries().size();
  32.             if (batchId == -1 || size == 0) { 
  33.                 // try { 
  34.                 // Thread.sleep(1000);
  35.                 // } catch (InterruptedException e) { 
  36.                 // }
  37.             } else { 
  38.                 synchronizedData(message.getEntries());
  39.             }
  41.             connector.ack(batchId); // 提交确认
  42.             // connector.rollback(batchId); // 处理失败, 回滚数据
  43.         }
  44.     }
  46.     /** * 同步数据 * @param entries * @throws Exception */
  47.     private static void synchronizedData(List<Entry> entries) throws Exception { 
  48.         for (Entry entry : entries) { 
  49.             if (entry.getEntryType() != EntryType.ROWDATA) { 
  50.                 continue;
  51.             }
  53.             RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
  54.             String tableName = entry.getHeader().getTableName();
  55.             for (RowData rowData : rowChange.getRowDatasList()) { 
  56.                 String sql = getSql(rowChange.getEventType(),tableName,rowData);
  57.                 System.out.println(sql);
  58.                 // TODO 执行sql语句
  59.             }
  60.         }
  61.     }
  63.     /** * 获取增删改的sql * @param eventType * @param tableName * @param rowData * @return */
  64.     private static String getSql(CanalEntry.EventType eventType,String tableName,RowData rowData){ 
  65.         String sql = null;
  66.         switch (eventType) { 
  67.             case INSERT:
  68.                 sql = getInsertSql(tableName,rowData.getAfterColumnsList());
  69.                 break;
  70.             case UPDATE:
  71.                 sql = getUpdateSql(tableName,rowData.getAfterColumnsList());
  72.                 break;
  73.             case DELETE:
  74.                 sql = getDeleteSql(tableName,rowData.getBeforeColumnsList());
  75.                 break;
  76.             default:
  77.                 break;
  78.         }
  79.         return sql;
  80.     }
  82.     private static String getInsertSql(String tableName,List<Column> columns){ 
  83.         if(columns.size() == 0 || StringUtils.isBlank(tableName)){ 
  84.             return null;
  85.         }
  86.         String keys = "";
  87.         String values = "";
  88.         for(int i=0;i<columns.size();i++){ 
  89.             if(i != 0) { 
  90.                 keys += ",";
  91.                 values += ",";
  92.             }
  93.             keys += columns.get(i).getName();
  94.             values += getValue(columns.get(i));
  95.         }
  96.         String format = "INSERT INTO %s (%s) VALUES (%s)";
  97.         return String.format(format,tableName,keys,values);
  98.     }
  100.     private static String getUpdateSql(String tableName,List<Column> columns){ 
  101.         if(columns.size() == 0 || StringUtils.isBlank(tableName)){ 
  102.             return null;
  103.         }
  104.         String sets = "";
  105.         String where = "";
  106.         for(Column column : columns){ 
  107.             if(column.getIsKey()){ 
  108.                 where = column.getName() + "=" + getValue(column);
  109.                 continue;
  110.             }
  111.             if(!StringUtils.isBlank(sets)) { 
  112.                 sets += ",";
  113.             }
  114.             sets += column.getName() + "=" + getValue(column);
  115.         }
  116.         String format = "UPDATE %s SET %s WHERE %s";
  117.         return String.format(format,tableName,sets,where);
  118.     }
  120.     private static String getDeleteSql(String tableName,List<Column> columns){ 
  121.         if(columns.size() == 0 || StringUtils.isBlank(tableName)){ 
  122.             return null;
  123.         }
  124.         String where = "";
  125.         for(Column column : columns){ 
  126.             if(column.getIsKey()){ 
  127.                 where = column.getName() + "=" + getValue(column);
  128.                 continue;
  129.             }
  130.         }
  131.         String format = "DELETE FROM %s WHERE %s";
  132.         return String.format(format,tableName,where);
  133.     }
  135.     private static String getValue(Column column){ 
  136.         if(column.getIsNull()){ 
  137.             return "null";
  138.         }
  139.         return String.format("'%s'",column.getValue());
  140.     }
  142. }

数据一致性

单机单点消费mysql的log-bin后直接更新到备份数据库中，数据一致性没有问题。但是如果变成分布式环境以及消费mysql的log-bin后将更新数据推到MQ中由多节点消费更新到多个备份数据库中，则会出现数据更新时序和数据一致性的问题。

而以上代码在update sql中除了获取值变化了的字段，也反查数据库获取了未变化的字段。因此每次update的sql实际上是该条记录的全量数据。

通过在表中加上时间戳字段作为记录的版本号，用逻辑删除取代物理删除delete，修改以上代码的sql拼接，insert操作时忽略主键冲突、update操作时仅更新版本号（时间戳）旧的记录，可以极大避免数据不一致的现象，也解决了MQ重复消费的问题。

  1. `last_update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP

再通过定时任务，每天一次增量数据更新，每周一次全量数据更新，保证数据的最终一致性。

发表评论

表情：
评论列表 （有 0 条评论，376人围观）

还没有评论，来说两句吧...

相关阅读