• canal简介

  • •        

      
      提供了另一种基于发布/订阅模式的同步机制，通过该框架我们可以对MySQL的binlog进行订阅，这样一旦MySQL中产生了新的写入、更新、删除等操作，就可以把binlog相关的消息推送至redis，redis在根据binlog中的记录，对redis进行更新。值得注意的是，MySQL的binlog需要手动打开，并且不会记录关于MySQL查询的命令和操作。
             其实这种机制很类似MySQL的主从备份机制，因为MySQL的主备也适合通过binlog来实现数据的一致性，而canal正是模仿了slave数据库的备份请求，使得redis的数据更新达到了相同的效果。
             binlog可以理解为一堆sql语言组成的日志。

       

• canal的使用

  •  canal的原理是基于mysql binlog技术，所以这里一定需要开启mysql的binlog写入功能，建议配置binlog模式为row.

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

     

  • canal的原理是模拟自己为mysql slave，所以这里一定需要做为mysql slave的相关权限.

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

    针对已有的账户可以直接通过grant授权。

  • canal安装和部署

    1. 访问：https://github.com/alibaba/canal/releases ，会列出所有历史的发布版本包 下载方式，比如以1.0.24版本为例子

       wget https://github.com/alibaba/canal/releases/download/canal-1.0.17/canal.deployer-1.0.17.tar.gz
       

        

    2. 解压缩

       canal.deployer-1.0.24.tar.gz

       mkdir /usr/local/canal
       -- 利用rz命令将压缩包复制到canal文件下，然后解压
       tar zxvf canal.deployer-1.0.24.tar.gz
       

        

    3. .解压完后进入/usr/local/canal目录，可以看到如下结构

       drwxr-xr-x
       2 root root 4096 Jan 30 05:28 bin/
       drwxr-xr-x
       4 root root 4096 Jan 30 01:34 conf/
       drwxr-xr-x
       2 root root 4096 Jan 30 01:34 lib/
       drwxrwxrwx
       4 root root 4096 Jan 30 01:38 logs/

        

    4. 配置修改

       vi conf/example/instance.properties

       #################################################
       ## mysql serverId
       canal.instance.mysql.slaveId = 1234
       #position info，需要改成自己的数据库信息 192.126.6.107为MySQL所在机器对应的IP
       canal.instance.master.address = 192.126.6.107: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 = ...
       #监听canal用户下以da_dosage_开头的表的变化
       #canal.instance.filter.regex = canal\.da_dosage_.*
       #################################################

        

    5. 启动

       -- cd到canal文件夹下的bin目录,运行以下命令启动canal服务
       ./startup.sh

        

    6. 查看日志

       vim logs/canal/canal.log
       ######################################
       2013-02-05 22:45:27.967 [main] INFO
       com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
       2013-02-05 22:45:28.113 [main] INFO
       com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.29.120:11111]
       2013-02-05 22:45:28.210 [main] INFO
       com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......
       ######################################

       vim logs/example/example.log
       ######################################
       2013-02-05 22:50:45.636 [main] INFO
       c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
       2013-02-05 22:50:45.641 [main] INFO
       c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
       2013-02-05 22:50:45.803 [main] INFO
       c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example
       2013-02-05 22:50:45.810 [main] INFO
       c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....
       ######################################

       如果日志如上，则说明启动canal成功，关闭canal的命令为

       -- cd到canal文件夹下的bin目录,运行以下命令停止canal服务
       ./stop.sh

        

• 利用canal同步MySQL和Redis的数据

  1. 添加依赖

     dependency>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.client</artifactId>
     <version>1.1.0</version>
     </dependency>

      

  2. 以下为canal客户端的代码

     import java.net.InetSocketAddress;
     import java.util.List;
     import com.alibaba.otter.canal.client.CanalConnectors;
     import com.alibaba.otter.canal.client.CanalConnector;
     import com.alibaba.otter.canal.common.utils.AddressUtils;
     import com.alibaba.otter.canal.protocol.Message;
     import com.alibaba.otter.canal.protocol.CanalEntry.Column;
     import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
     import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
     import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
     import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
     import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
     public class SimpleCanalClientExample {
     public static void main(String args[]) {
     // 创建链接
     AddressUtils.getHostIp()为canal所在服务器对应的IP
     CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
     11111), "example", "", "");
     int batchSize = 1000;
     int emptyCount = 0;
     try {
     connector.connect();
     connector.subscribe(".*\..*");
     // connector.subscribe(".*\.da_dosage_.*"); 监听以da_dasage_开头的表名的变化，其他表的变化将不被canal监听
     connector.rollback();
     int totalEmptyCount = 120;
     while (emptyCount < totalEmptyCount) {
     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) {
     }
     } 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<Entry> 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("================&gt; 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) {
     RedisDB.delObject(key,field);
     printColumn(rowData.getBeforeColumnsList()); // 打印被删除的数据
     } else if (eventType == EventType.INSERT) {
     RedisDB.putObject(key,field,JSONObject.toJSONString(value));
     printColumn(rowData.getAfterColumnsList()); // 打印新增的数据
     } else {
     RedisDB.putObject(key,field,JSONObject.toJSONString(value));
     System.out.println("-------&gt; before");
     printColumn(rowData.getBeforeColumnsList()); // 打印修改前的数据
     System.out.println("-------&gt; after");
     printColumn(rowData.getAfterColumnsList()); // 打印修改后的数据
     }
     }
     }
     }
     private static void printColumn(List<Column> columns) {
     for (Column column : columns) {
     System.out.println(column.getName() + " : " + column.getValue() + "
     update=" + column.getUpdated());
     }
     }

  3. RedisDB.java

     package com.common.util;
     import java.io.FileNotFoundException;
     import java.io.IOException;
     import java.io.InputStream;
     import java.io.UnsupportedEncodingException;
     import java.util.Iterator;
     import java.util.LinkedHashMap;
     import java.util.Map;
     import java.util.Properties;
     import java.util.Set;
     import java.util.Map.Entry;
     import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
     import org.slf4j.Logger;
     import org.slf4j.LoggerFactory;
     import redis.clients.jedis.Jedis;
     import redis.clients.jedis.JedisPool;
     public class RedisDB{
     private static final Logger logger = LoggerFactory.getLogger(RedisDB.class);
     public static Map<String,String> propMap = new LinkedHashMap<String,String>();
     static{
     try{
     InputStream inputStream = RedisDB.class.getClassLoader().getResourceAsStream("application.properties");
     Properties prop = new Properties();
     prop.load(inputStream);
     Iterator<Entry<Object,Object>> it = prop.entrySet().iterator();
     while(it.hasNext()){
     Entry<Object,Object> entry = it.next();
     Object key = entry.getKey();
     Object value = entry.getValue();
     propMap.put(key.toString(),value.toString());
     }
     }catch(FileNotFoundException e){
     System.out.println("文件没有找到！");
     }catch(IOException e){
     e.printStackTrace();
     }
     }
     /* 最大连接数 */
     private static int maxTotal = Integer.parseInt(propMap.get("redis.maxTotal"));
     /* 最大空闲等待数 */
     private static int maxIdle = Integer.parseInt(propMap.get("redis.maxIdle"));
     /* 最大等待时间 */
     private static int maxWaitMillis = Integer.parseInt(propMap.get("redis.maxWaitMillis"));
     /* 从pool中获取连接时，是否检查连接可用 */
     private static boolean testOnBorrow = Boolean.parseBoolean(propMap.get("redis.testOnBorrow"));
     /* 端口号 */
     private static int port = Integer.parseInt(propMap.get("redis.port"));
     /* ip地址 */
     private static String hostName = propMap.get("redis.hostName");
     /* redis连接密码 */
     private static String password = propMap.get("redis.password");
     /* 是否对空闲连接对象进行检查 */
     private static boolean testOnIdle = propMap.get("redis.testOnIdle").equalsIgnoreCase("true")?true:false;
     /* 每隔多少秒检查一次空闲连接对象 */
     private static int timeBetweenEvictionRunsMillis = Integer.valueOf(propMap.get("redis.timeBetweenEvictionRunsMillis"));
     /* 一次驱逐过程中最多驱逐对象的个数 */
     private static int numTestsPerEvictionRun = Integer.parseInt(propMap.get("redis.numTestsPerEvictionRun"));
     /*
     * 表示一个对象至少停留在idle状态的最短时间，然后才能被idle object
     * evitor扫描并驱逐；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义
     */
     private static int minEvictableIdleTimeMillis = Integer.parseInt(propMap.get("redis.minEvictableIdleTimeMillis"));
     /* 连接超时 */
     private static int timeout = Integer.parseInt(propMap.get("redis.timeout"));
     private static int db = Integer.parseInt(propMap.get("redis.dbSelected"));
     /* jedis连接池对象 */
     private static JedisPool jedisPool = null;
     private static Object lock = new Object();
     public static void init(){
     if(null==jedisPool){
     GenericObjectPoolConfig config = new GenericObjectPoolConfig();
     config.setMaxTotal(maxTotal);
     config.setMaxIdle(maxIdle);
     config.setMaxWaitMillis(maxWaitMillis);
     config.setTestOnBorrow(testOnBorrow);
     config.setTestWhileIdle(testOnIdle);
     config.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis);
     config.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);
     config.setNumTestsPerEvictionRun(numTestsPerEvictionRun);
     synchronized(lock){
     try{
     jedisPool = new JedisPool(config,hostName,port,timeout,password);
     logger.debug("init jedis pool success!");
     }catch(Exception e){
     logger.debug("init jedis pool fail!"+e.getMessage());
     }
     }
     }
     }
     public static Jedis getJedisConn(){
     if(null==jedisPool){
     init();
     }
     Jedis retJedisConnection = null;
     try{
     retJedisConnection = jedisPool.getResource();
     }catch(Exception e){
     logger.error(e.getMessage());
     jedisPool = null;
     init();
     retJedisConnection = jedisPool.getResource();
     }
     return retJedisConnection;
     }
     public static void closeJedis(Jedis jedis){
     if(null!=jedis){
     jedis.close();
     }
     }
     public static boolean isExists(String key,String field){
     Jedis jedis = null;
     boolean b = false;
     try{
     jedis = RedisDB.getJedisConn();
     jedis.select(db);
     b = jedis.hexists(key,field);
     }catch(Exception e){
     logger.error(e.getMessage());
     }finally{
     RedisDB.closeJedis(jedis);
     }
     return b;
     }
     public static String getObject(String key,String field){
     String values = null;
     Jedis jedis = null;
     try{
     jedis = RedisDB.getJedisConn();
     jedis.select(db);
     values = jedis.hget(key,field);
     }catch(Exception e){
     logger.error(e.getMessage());
     }finally{
     RedisDB.closeJedis(jedis);
     }
     return values;
     }
     public static void putObject(String key,String field,String value){
     Jedis jedis = null;
     try{
     jedis = RedisDB.getJedisConn();
     jedis.select(db);
     jedis.hset(key,field,value);
     }catch(Exception e){
     e.printStackTrace();
     logger.error(e.getMessage());
     }finally{
     RedisDB.closeJedis(jedis);
     }
     }
     public static void delObject(String key,String field){
     Jedis jedis = null;
     try{
     jedis = RedisDB.getJedisConn();
     jedis.select(db);
     if(isExists(key,field)){
     jedis.hdel(key,field);
     }
     }catch(Exception e){
     e.printStackTrace();
     logger.error(e.getMessage());
     }finally{
     RedisDB.closeJedis(jedis);
     }
     }
     public static void main(String[] args) throws UnsupportedEncodingException{
     Jedis jedis = getJedisConn();
     Set<String> keys = jedis.keys("*");
     System.out.println(keys);
     for(String string:keys){
     System.out.println(string);
     }
     Set<byte[]> keys2 = jedis.keys("*".getBytes("utf-8"));
     for(byte[] bs:keys2){
     System.out.println(bs.length);
     System.out.println(new String(bs));
     }
     }
     }
     

      

  4. 启动Canal Client后，可以从控制台从看到类似消息：
     empty count : 1
     empty count : 2
     empty count : 3
     empty count : 4
     此时代表当前数据库无变更数据
     触发数据库变更
     mysql> use test;
     Database changed
     mysql> CREATE TABLE `xdual` (
     ->
     `ID` int(11) NOT NULL AUTO_INCREMENT,
     ->
     `X` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
     ->
     PRIMARY KEY (`ID`)
     -> ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 ;
     Query OK, 0 rows affected (0.06 sec)
     mysql> insert into xdual(id,x) values(null,now());Query OK, 1 row affected (0.06 sec)
     可以从控制台中看到：
     empty count : 1
     empty count : 2
     empty count : 3
     empty count : 4
     ================> binlog[mysql-bin.001946:313661577] , name[test,xdual] , eventType : INSERT
     ID : 4
     update=true
     X : 2013-02-05 23:29:46
     update=true

     这样就能同步MySQL和Redis的数据啦~~~

  5. application.properties

     server.port=8080
     #server.servlet.context-path=/demo
     #集成jsp
     spring.mvc.view.prefix=/WEB-INF/views/
     spring.mvc.view.suffix=.jsp
     #集成mybatis：注解的形式
     spring.datasource.driverClassName=com.mysql.jdbc.Driver
     spring.datasource.url=jdbc:mysql://127.0.0.1:3306/NEEM_CDB?useUnicode=true&characterEncoding=utf-8&useSSL=false
     spring.datasource.username=root
     spring.datasource.password=root
     #mybatis配置信息
     mybatis.type-aliases-package=com.data.pojo
     #集成mybatis：配置xml形式
     mybatis.config-locations=classpath:mybatis/mybatis-config.xml
     mybatis.mapper-locations=classpath:mybatis/mapper/*.xml
     #集成redis
     #redis数据库参数
     #jedis连接池配置
     redis.maxTotal=100
     redis.maxIdle=5
     redis.minIdle=2
     redis.maxWaitMillis=30000
     redis.testOnBorrow=false
     redis.testOnIdle=true
     redis.timeBetweenEvictionRunsMillis=60000
     redis.numTestsPerEvictionRun=1000
     redis.minEvictableIdleTimeMillis=3000000
     redis.port=6379
     redis.hostName=20.20.20.246
     redis.password=123456
     redis.timeout=10000
     #选择redis数据库
     redis.dbSelected=15
     #打印sql
     #logging.level.com.data.dao=debug
     #################################################日志####################################################
     #com.mycompany.mavenspringboot.controller 日志 WARN级别输出
     #logging.level.com.mycompany.mavenspringboot.controller=WARN
     #com.mycompany.mavenspringboot.mapper sql日志 DEBUG级别输出
     #logging.level.com=DEBUG,INFO
     #logging.file=logs/statistical.log
     #logging.pattern.console=%d{yyyy/MM/dd-HH:mm:ss} [%thread] %-5level %logger- %msg%n
     #logging.pattern.file=%d{yyyy/MM/dd-HH:mm} [%thread] %-5level %logger- %msg%n
     

      

  6. logback.xml

     <?xml version="1.0" encoding="UTF-8"?>
     <configuration >
     <!-- 彩色日志 -->
     <!-- 彩色日志依赖的渲染类 -->
     <conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter" />
     <conversionRule conversionWord="wex" converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter" />
     <conversionRule conversionWord="wEx" converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter" />
     <!-- 彩色日志格式 -->
     <property name="CONSOLE_LOG_PATTERN" value="${CONSOLE_LOG_PATTERN:-%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%logger){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}" />
     <!--包名输出缩进对齐-->
     <property name="CONSOLE_LOG_PATTERN" value="${CONSOLE_LOG_PATTERN:-%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}" />
     <contextName>fanxlxs</contextName>
     <!--文件夹在当前项目磁盘根目录-->
     <!--
     <property name="LOG_PATH" value="/modellog" /> -->
     <!--设置系统日志目录-->
     <property name="APPDIR" value="logs" />
     <!--
     日志记录器，日期滚动记录
     ERROR 级别
     -->
     <appender name="ERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
     <!-- 正在记录的日志文件的路径及文件名 -->
     <file>${catalina.home}/${APPDIR}/statisticalData_error.log</file>
     <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
     <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
     <!-- 归档的日志文件的路径，例如今天是1992-11-06日志，当前写的日志文件路径为file节点指定，
     可以将此文件与file指定文件路径设置为不同路径，从而将当前日志文件或归档日志文件置不同的目录。
     而1992-11-06的日志文件在由fileNamePattern指定。%d{yyyy-MM-dd}指定日期格式，%i指定索引 -->
     <fileNamePattern>${catalina.home}/${APPDIR}/error/statisticalData-error-%d{yyyy-MM-dd}.%i.zip</fileNamePattern>
     <!-- 除按日志记录之外，还配置了日志文件不能超过10MB，若超过10MB，日志文件会以索引0开始，
     命名日志文件，例如log-error-1992-11-06.0.log -->
     <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
     <maxFileSize>10MB</maxFileSize>
     </timeBasedFileNamingAndTriggeringPolicy>
     </rollingPolicy>
     <!-- 追加方式记录日志 -->
     <append>true</append>
     <!-- 日志文件的格式 -->
     <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
     <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level --- [%thread] %logger Line:%-3L - %msg%n</pattern>
     <charset>utf-8</charset>
     </encoder>
     <!-- 此日志文件记录error级别的 -->
     <filter class="ch.qos.logback.classic.filter.LevelFilter">
     <level>error</level>
     <onMatch>ACCEPT</onMatch>
     <onMismatch>DENY</onMismatch>
     </filter>
     </appender>
     <!-- 日志记录器，日期滚动记录
     WARN
     级别
     -->
     <appender name="WARN" class="ch.qos.logback.core.rolling.RollingFileAppender">
     <!-- 正在记录的日志文件的路径及文件名 -->
     <file>${catalina.home}/${APPDIR}/statisticalData_warn.log</file>
     <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
     <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
     <!-- 归档的日志文件的路径，例如今天1992-11-06日志，当前写的日志文件路径为file节点指定，
     可以将此文件与file指定文件路径设置为不同路径，从而将当前日志文件或归档日志文件置不同的目录。
     而1992-11-06的日志文件在由fileNamePattern指定。%d{yyyy-MM-dd}指定日期格式，%i指定索引 -->
     <fileNamePattern>${catalina.home}/${APPDIR}/warn/statisticalData-warn-%d{yyyy-MM-dd}.%i.zip</fileNamePattern>
     <!-- 除按日志记录之外，还配置了日志文件不能超过10MB，若超过10MB，日志文件会以索引0开始，
     命名日志文件，例如log-warn-1992-11-06.0.log -->
     <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
     <maxFileSize>10MB</maxFileSize>
     </timeBasedFileNamingAndTriggeringPolicy>
     </rollingPolicy>
     <!-- 追加方式记录日志 -->
     <append>true</append>
     <!-- 日志文件的格式 -->
     <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
     <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level --- [%thread] %logger Line:%-3L - %msg%n</pattern>
     <charset>utf-8</charset>
     </encoder>
     <!-- 此日志文件只记录warn级别的 -->
     <filter class="ch.qos.logback.classic.filter.LevelFilter">
     <level>warn</level>
     <onMatch>ACCEPT</onMatch>
     <onMismatch>DENY</onMismatch>
     </filter>
     </appender>
     <!-- 日志记录器，日期滚动记录
     INFO
     级别
     -->
     <appender name="INFO" class="ch.qos.logback.core.rolling.RollingFileAppender">
     <!-- 正在记录的日志文件的路径及文件名 -->
     <file>${catalina.home}/${APPDIR}/statisticalData_info.log</file>
     <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
     <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
     <!-- 归档的日志文件的路径，例如今天是1992-11-06日志，当前写的日志文件路径为file节点指定，
     可以将此文件与file指定文件路径设置为不同路径，从而将当前日志文件或归档日志文件置不同的目录。
     而1992-11-06的日志文件在由fileNamePattern指定。%d{yyyy-MM-dd}指定日期格式，%i指定索引 -->
     <fileNamePattern>${catalina.home}/${APPDIR}/info/statisticalData-info-%d{yyyy-MM-dd}.%i.zip</fileNamePattern>
     <!-- 除按日志记录之外，还配置了日志文件不能超过10MB，若超过10MB，日志文件会以索引0开始，
     命名日志文件，例如log-info-1992-11-06.0.log -->
     <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
     <maxFileSize>10MB</maxFileSize>
     </timeBasedFileNamingAndTriggeringPolicy>
     </rollingPolicy>
     <!-- 追加方式记录日志 -->
     <append>true</append>
     <!-- 日志文件的格式 -->
     <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
     <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level --- [%thread] %logger Line:%-3L - %msg%n</pattern>
     <charset>utf-8</charset>
     </encoder>
     <!-- 此日志文件只记录info级别的 -->
     <filter class="ch.qos.logback.classic.filter.LevelFilter">
     <level>info</level>
     <onMatch>ACCEPT</onMatch>
     <onMismatch>DENY</onMismatch>
     </filter>
     </appender>
     <!-- 日志记录器，日期滚动记录
     DEBUG
     级别
     -->
     <appender name="DEBUG" class="ch.qos.logback.core.rolling.RollingFileAppender">
     <!-- 正在记录的日志文件的路径及文件名 -->
     <file>${catalina.home}/${APPDIR}/statisticalData_debug.log</file>
     <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
     <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
     <!-- 归档的日志文件的路径，例如今天是1992-11-06日志，当前写的日志文件路径为file节点指定，
     可以将此文件与file指定文件路径设置为不同路径，从而将当前日志文件或归档日志文件置不同的目录。
     而1992-11-06的日志文件在由fileNamePattern指定。%d{yyyy-MM-dd}指定日期格式，%i指定索引 -->
     <fileNamePattern>${catalina.home}/${APPDIR}/debug/statisticalData-debug-%d{yyyy-MM-dd}.%i.zip</fileNamePattern>
     <!-- 除按日志记录之外，还配置了日志文件不能超过10MB，若超过10MB，日志文件会以索引0开始，
     命名日志文件，例如log-debug-1992-11-06.0.log -->
     <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
     <maxFileSize>10MB</maxFileSize>
     </timeBasedFileNamingAndTriggeringPolicy>
     </rollingPolicy>
     <!-- 追加方式记录日志 -->
     <append>true</append>
     <!-- 日志文件的格式 -->
     <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
     <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level --- [%thread] %logger Line:%-3L - %msg%n</pattern>
     <charset>utf-8</charset>
     </encoder>
     <!-- 此日志文件只记录debug级别的 -->
     <filter class="ch.qos.logback.classic.filter.LevelFilter">
     <level>debug</level>
     <onMatch>ACCEPT</onMatch>
     <onMismatch>DENY</onMismatch>
     </filter>
     </appender>
     <!-- ConsoleAppender 控制台输出日志 -->
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
     <!--encoder 默认配置为PatternLayoutEncoder-->
     <encoder>
     <pattern>${CONSOLE_LOG_PATTERN}</pattern>
     <!--<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level -&#45;&#45; [%thread] %logger Line:%-3L - %msg%n</pattern>-->
     <charset>utf-8</charset>
     </encoder>
     <!--此日志appender是为开发使用，只配置最底级别，控制台输出的日志级别是大于或等于此级别的日志信息-->
     <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
     <level>debug</level>
     </filter>
     </appender>
     <!-- FrameworkServlet日志-->
     <logger name="org.springframework" level="WARN" />
     <!-- mybatis日志打印-->
     <logger name="org.apache.ibatis" level="DEBUG" />
     <logger name="java.sql" level="DEBUG" />
     <!--
     项目 mapper 路径
     console控制台显示sql语句：STDOUT.filter.level -> debug级别
     -->
     <logger name="com.common.dao" level="DEBUG"></logger>
     <logger name="com.data.dao" level="DEBUG"></logger>
     <logger name="com.data_redis.dao" level="DEBUG"></logger>
     <!-- 生产环境下，将此级别配置为适合的级别，以免日志文件太多或影响程序性能 -->
     <root level="INFO">
     <appender-ref ref="ERROR" />
     <appender-ref ref="WARN" />
     <appender-ref ref="INFO" />
     <appender-ref ref="DEBUG" />
     <!-- 生产环境将请stdout去掉 -->
     <appender-ref ref="STDOUT" />
     </root>
     </configuration>

     另附canal配置文件详解：https://blog.csdn.net/xiaolong_4_2/article/details/85071112 

最后

以上就是饱满毛豆为你收集整理的阿里巴巴中间件canal介绍和利用canal同步MySQL和Redis数据canal简介canal的使用利用canal同步MySQL和Redis的数据的全部内容，希望文章能够帮你解决阿里巴巴中间件canal介绍和利用canal同步MySQL和Redis数据canal简介canal的使用利用canal同步MySQL和Redis的数据所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。