当我们需要对数据库的修改做监听的时候，可以使用阿里的canal。例如，可以实现将mysql中修改的数据更新到redis中，而不用每个service都分两步去先操作mysql再操作redis。

配置mysql

启用mysql的binlog

# vim /etc/my.cnf添加：[mysqld]  log-bin=mysql-bin #添加这一行就ok  binlog-format=ROW #选择row模式  server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复

添加canal数据库用户

CREATE USER canal IDENTIFIED BY 'canal';      GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';    -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;    FLUSH PRIVILEGES;

配置canal

解压

mkdir /tmp/canal  tar zxvf canal.deployer-$version.tar.gz  -C /tmp/canal

修改配置文件

vi conf/example/instance.properties  #################################################  ## mysql serverId  canal.instance.mysql.slaveId = 1234    # position info，需要改成自己的数据库信息  canal.instance.master.address = 127.0.0.1:3306   canal.instance.master.journal.name =   canal.instance.master.position =   canal.instance.master.timestamp =     #canal.instance.standby.address =   #canal.instance.standby.journal.name =  #canal.instance.standby.position =   #canal.instance.standby.timestamp =     # username/password，需要改成自己的数据库信息  canal.instance.dbUsername = canal    canal.instance.dbPassword = canal  canal.instance.defaultDatabaseName =  canal.instance.connectionCharset = UTF-8    # table regex  canal.instance.filter.regex = .*\\..*    #################################################

启动

sh bin/startup.sh

public class ClientSample {      public static void main(String args[]) {          // 创建链接          CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("10.0.0.2",                11111), "example", "", "");          int batchSize = 1000;          int emptyCount = 0;          try {              connector.connect();              connector.subscribe(".*\\..*");              connector.rollback();              int totalEmtryCount = 1200;              while (emptyCount < totalEmtryCount) {                  Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据                long batchId = message.getId();                  int size = message.getEntries().size();                  if (batchId == -1 || size == 0) {                      emptyCount++;                      System.out.println("empty count : " + emptyCount);                      try {                          Thread.sleep(1000);                      } catch (InterruptedException e) {                          e.printStackTrace();                      }                  } else {                      emptyCount = 0;                      // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);                      printEntry(message.getEntries());                  }                    connector.ack(batchId); // 提交确认                  // connector.rollback(batchId); // 处理失败, 回滚数据              }                System.out.println("empty too many times, exit");          } finally {              connector.disconnect();          }      }        private static void printEntry(List
    
      entrys) {        for (Entry entry : entrys) {              if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {                  continue;              }                RowChange rowChage = null;              try {                  rowChage = RowChange.parseFrom(entry.getStoreValue());              } catch (Exception e) {                  throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),                          e);              }                EventType eventType = rowChage.getEventType();              System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",                      entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),                      entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),                      eventType));                for (RowData rowData : rowChage.getRowDatasList()) {                  if (eventType == EventType.DELETE) {                      printColumn(rowData.getBeforeColumnsList());                    //redisDelete(rowData.getBeforeColumnsList());                } else if (eventType == EventType.INSERT) {                      printColumn(rowData.getAfterColumnsList());                  } else {                      System.out.println("-------> before");                    //修改之前每个字段的值                    printColumn(rowData.getBeforeColumnsList());                      System.out.println("-------> after");                    //修改之后每个字段的值                    printColumn(rowData.getAfterColumnsList());                  }              }          }      }        private static void printColumn(List
     
       columns) {        for (Column column : columns) {            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());        }      }  }
     
    

注意

canal每次会记录客户端消费的位置。如果canal挂掉，mysql在不断更新，redis更新滞后。重启canal后，会自动同步redis更新。

HA

server端可参考官方文档

client也是靠zookeeper实现ha

CanalConnector connector = CanalConnectors.newClusterConnector("192.168.213.44:4180", "example", "", "");  //client连接zookeeper地址          int batchSize = 1;          int emptyCount = 0;          //启动多个实例，不断尝试连接zookeeper实现HA。        while(true) {              try {                  connector.connect();                  connector.subscribe(".*\\..*");                  ...        }

Reference