zoukankan      html  css  js  c++  java

  • 利用Canal完成Mysql数据同步Redis

    流程
    Canal的原理是模拟Slave向Master发送请求，Canal解析binlog，但不将解析结果持久化，而是保存在内存中，每次有客户端读取一次消息，就删除该消息。这里所说的客户端，就需要我们写一个连接Canal的程序，持续从Canal获取数据。

    步骤
    一、配置Canal
    参考https://github.com/alibaba/canal

    【mysql配置】
    1，配置参数

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

    2，在mysql中 配置canal数据库管理用户，配置相应权限（repication权限）

    [html] view plain copy
     
    1. CREATE USER canal IDENTIFIED BY 'canal';      
    2. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';    
    3. -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;    
    4. FLUSH PRIVILEGES;    

    【canal下载和配置】
    1，下载canal https://github.com/alibaba/canal/releases  
    2，解压

    [html] view plain copy
     
    1. mkdir /tmp/canal  
    2. tar zxvf canal.deployer-$version.tar.gz  -C /tmp/canal  

    3，修改配置文件

    [html] view plain copy
     
    1. vi conf/example/instance.properties  
    [html] view plain copy
     
    1. #################################################  
    2. ## mysql serverId  
    3. canal.instance.mysql.slaveId = 1234  
    4.   
    5. # position info，需要改成自己的数据库信息  
    6. canal.instance.master.address = 127.0.0.1:3306   
    7. canal.instance.master.journal.name =   
    8. canal.instance.master.position =   
    9. canal.instance.master.timestamp =   
    10.   
    11. #canal.instance.standby.address =   
    12. #canal.instance.standby.journal.name =  
    13. #canal.instance.standby.position =   
    14. #canal.instance.standby.timestamp =   
    15.   
    16. # username/password，需要改成自己的数据库信息  
    17. canal.instance.dbUsername = canal    
    18. canal.instance.dbPassword = canal  
    19. canal.instance.defaultDatabaseName =  
    20. canal.instance.connectionCharset = UTF-8  
    21.   
    22. # table regex  
    23. canal.instance.filter.regex = .*\..*  
    24.   
    25. #################################################  


    【canal启动和关闭】
    1，启动

    [html] view plain copy
     
    1. sh bin/startup.sh  

    2，查看日志

    [html] view plain copy
     
    1. vi logs/canal/canal.log  
    [html] view plain copy
     
    1. 2013-02-05 22:45:27.967 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.  
    2. <pre name="user-content-code">2013-02-05 22:45:28.113 [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.29.120:11111]  
    3. 2013-02-05 22:45:28.210 [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......  

    具体instance的日志：

    [html] view plain copy
     
    1. vi logs/example/example.log  
    [html] view plain copy
     
    1. 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]  
    2. 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]  
    3. 2013-02-05 22:50:45.803 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example   
    4. 2013-02-05 22:50:45.810 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....  

    3，关闭

    [html] view plain copy
     
    1. sh bin/stop.sh  


    注意：
    1，这里只需要配置好参数后，就可以直接运行
    2，Canal没有解析后的文件，不会持久化

    二、创建客户端
    参考https://github.com/alibaba/canal/wiki/ClientExample


    其中一个是连接canal并操作的类，一个是redis的工具类，使用maven主要是依赖包的下载很方便。

    pom.xml

    [objc] view plain copy
     
    1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  
    2.   <modelVersion>4.0.0</modelVersion>  
    3.   <groupId>com.alibaba.otter</groupId>  
    4.   <artifactId>canal.sample</artifactId>  
    5.   <version>0.0.1-SNAPSHOT</version>  
    6.   <dependencies>  
    7.     <dependency>    
    8.         <groupId>com.alibaba.otter</groupId>    
    9.         <artifactId>canal.client</artifactId>    
    10.         <version>1.0.12</version>    
    11.     </dependency>    
    12.       
    13.     <dependency>    
    14.         <groupId>org.springframework</groupId>    
    15.         <artifactId>spring-test</artifactId>    
    16.         <version>3.1.2.RELEASE</version>    
    17.         <scope>test</scope>    
    18.     </dependency>    
    19.         
    20.     <dependency>    
    21.         <groupId>redis.clients</groupId>    
    22.         <artifactId>jedis</artifactId>    
    23.         <version>2.4.2</version>    
    24.     </dependency>    
    25.       
    26.     </dependencies>  
    27.   <build/>  
    28. </project>  




    2，ClientSample代码
    这里主要做两个工作，一个是循环从Canal上取数据，一个是将数据更新至Redis

    [cpp] view plain copy
     
    1. package canal.sample;  
    2.   
    3. import java.net.InetSocketAddress;    
    4. import java.util.List;    
    5.   
    6. import com.alibaba.fastjson.JSONObject;  
    7. import com.alibaba.otter.canal.client.CanalConnector;    
    8. import com.alibaba.otter.canal.common.utils.AddressUtils;    
    9. import com.alibaba.otter.canal.protocol.Message;    
    10. import com.alibaba.otter.canal.protocol.CanalEntry.Column;    
    11. import com.alibaba.otter.canal.protocol.CanalEntry.Entry;    
    12. import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;    
    13. import com.alibaba.otter.canal.protocol.CanalEntry.EventType;    
    14. import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;    
    15. import com.alibaba.otter.canal.protocol.CanalEntry.RowData;    
    16. import com.alibaba.otter.canal.client.*;    
    17.    
    18. public class ClientSample {    
    19.   
    20.    public static void main(String args[]) {    
    21.          
    22.        // 创建链接    
    23.        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),    
    24.                11111), "example", "", "");    
    25.        int batchSize = 1000;    
    26.        try {    
    27.            connector.connect();    
    28.            connector.subscribe(".*\..*");    
    29.            connector.rollback();      
    30.            while (true) {    
    31.                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据    
    32.                long batchId = message.getId();    
    33.                int size = message.getEntries().size();    
    34.                if (batchId == -1 || size == 0) {    
    35.                    try {    
    36.                        Thread.sleep(1000);    
    37.                    } catch (InterruptedException e) {    
    38.                        e.printStackTrace();    
    39.                    }    
    40.                } else {    
    41.                    printEntry(message.getEntries());    
    42.                }    
    43.    
    44.                connector.ack(batchId); // 提交确认    
    45.                // connector.rollback(batchId); // 处理失败, 回滚数据    
    46.            }    
    47.    
    48.        } finally {    
    49.            connector.disconnect();    
    50.        }    
    51.    }    
    52.    
    53.    private static void printEntry( List<Entry> entrys) {    
    54.        for (Entry entry : entrys) {    
    55.            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {    
    56.                continue;    
    57.            }    
    58.    
    59.            RowChange rowChage = null;    
    60.            try {    
    61.                rowChage = RowChange.parseFrom(entry.getStoreValue());    
    62.            } catch (Exception e) {    
    63.                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),    
    64.                        e);    
    65.            }    
    66.    
    67.            EventType eventType = rowChage.getEventType();    
    68.            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",    
    69.                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),    
    70.                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),    
    71.                    eventType));    
    72.    
    73.            for (RowData rowData : rowChage.getRowDatasList()) {    
    74.                if (eventType == EventType.DELETE) {    
    75.                    redisDelete(rowData.getBeforeColumnsList());    
    76.                } else if (eventType == EventType.INSERT) {    
    77.                    redisInsert(rowData.getAfterColumnsList());    
    78.                } else {    
    79.                    System.out.println("-------> before");    
    80.                    printColumn(rowData.getBeforeColumnsList());    
    81.                    System.out.println("-------> after");    
    82.                    redisUpdate(rowData.getAfterColumnsList());    
    83.                }    
    84.            }    
    85.        }    
    86.    }    
    87.    
    88.    private static void printColumn( List<Column> columns) {    
    89.        for (Column column : columns) {    
    90.            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());    
    91.        }    
    92.    }    
    93.      
    94.       private static void redisInsert( List<Column> columns){  
    95.           JSONObject json=new JSONObject();  
    96.           for (Column column : columns) {    
    97.               json.put(column.getName(), column.getValue());    
    98.            }    
    99.           if(columns.size()>0){  
    100.               RedisUtil.stringSet("user:"+ columns.get(0).getValue(),json.toJSONString());  
    101.           }  
    102.        }  
    103.         
    104.       private static  void redisUpdate( List<Column> columns){  
    105.           JSONObject json=new JSONObject();  
    106.           for (Column column : columns) {    
    107.               json.put(column.getName(), column.getValue());    
    108.            }    
    109.           if(columns.size()>0){  
    110.               RedisUtil.stringSet("user:"+ columns.get(0).getValue(),json.toJSONString());  
    111.           }  
    112.       }  
    113.     
    114.        private static  void redisDelete( List<Column> columns){  
    115.            JSONObject json=new JSONObject();  
    116.               for (Column column : columns) {    
    117.                   json.put(column.getName(), column.getValue());    
    118.                }    
    119.               if(columns.size()>0){  
    120.                   RedisUtil.delKey("user:"+ columns.get(0).getValue());  
    121.               }  
    122.        }  
    123.   
    124.      
    125. }    


    3，RedisUtil代码

    [objc] view plain copy
     
    1. package canal.sample;  
    2.   
    3. import redis.clients.jedis.Jedis;  
    4. import redis.clients.jedis.JedisPool;  
    5. import redis.clients.jedis.JedisPoolConfig;  
    6.   
    7. public class RedisUtil {  
    8.   
    9.     // Redis服务器IP  
    10.     private static String ADDR = "10.1.2.190";  
    11.   
    12.     // Redis的端口号  
    13.     private static int PORT = 6379;  
    14.   
    15.     // 访问密码  
    16.     private static String AUTH = "admin";  
    17.   
    18.     // 可用连接实例的最大数目，默认值为8；  
    19.     // 如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis实例，则此时pool的状态为exhausted(耗尽)。  
    20.     private static int MAX_ACTIVE = 1024;  
    21.   
    22.     // 控制一个pool最多有多少个状态为idle(空闲的)的jedis实例，默认值也是8。  
    23.     private static int MAX_IDLE = 200;  
    24.   
    25.     // 等待可用连接的最大时间，单位毫秒，默认值为-1，表示永不超时。如果超过等待时间，则直接抛出JedisConnectionException；  
    26.     private static int MAX_WAIT = 10000;  
    27.   
    28.     // 过期时间  
    29.     protected static int  expireTime = 660 * 660 *24;  
    30.       
    31.     // 连接池  
    32.     protected static JedisPool pool;  
    33.   
    34.     /** 
    35.      * 静态代码，只在初次调用一次 
    36.      */  
    37.     static {  
    38.         JedisPoolConfig config = new JedisPoolConfig();  
    39.         //最大连接数  
    40.         config.setMaxTotal(MAX_ACTIVE);  
    41.         //最多空闲实例  
    42.         config.setMaxIdle(MAX_IDLE);  
    43.         //超时时间  
    44.         config.setMaxWaitMillis(MAX_WAIT);  
    45.         //  
    46.         config.setTestOnBorrow(false);  
    47.         pool = new JedisPool(config, ADDR, PORT, 1000);  
    48.     }  
    49.   
    50.     /** 
    51.      * 获取jedis实例 
    52.      */  
    53.     protected static synchronized Jedis getJedis() {  
    54.         Jedis jedis = null;  
    55.         try {  
    56.             jedis = pool.getResource();  
    57.         } catch (Exception e) {  
    58.             e.printStackTrace();  
    59.             if (jedis != null) {  
    60.                 pool.returnBrokenResource(jedis);  
    61.             }  
    62.         }  
    63.         return jedis;  
    64.     }  
    65.   
    66.     /** 
    67.      * 释放jedis资源 
    68.      *  
    69.      * @param jedis 
    70.      * @param isBroken 
    71.      */  
    72.     protected static void closeResource(Jedis jedis, boolean isBroken) {  
    73.         try {  
    74.             if (isBroken) {  
    75.                 pool.returnBrokenResource(jedis);  
    76.             } else {  
    77.                 pool.returnResource(jedis);  
    78.             }  
    79.         } catch (Exception e) {  
    80.   
    81.         }  
    82.     }  
    83.   
    84.     /** 
    85.      *  是否存在key 
    86.      *  
    87.      * @param key 
    88.      */  
    89.     public static boolean existKey(String key) {  
    90.         Jedis jedis = null;  
    91.         boolean isBroken = false;  
    92.         try {  
    93.             jedis = getJedis();  
    94.             jedis.select(0);  
    95.             return jedis.exists(key);  
    96.         } catch (Exception e) {  
    97.             isBroken = true;  
    98.         } finally {  
    99.             closeResource(jedis, isBroken);  
    100.         }  
    101.         return false;  
    102.     }  
    103.   
    104.     /** 
    105.      *  删除key 
    106.      *  
    107.      * @param key 
    108.      */  
    109.     public static void delKey(String key) {  
    110.         Jedis jedis = null;  
    111.         boolean isBroken = false;  
    112.         try {  
    113.             jedis = getJedis();  
    114.             jedis.select(0);  
    115.             jedis.del(key);  
    116.         } catch (Exception e) {  
    117.             isBroken = true;  
    118.         } finally {  
    119.             closeResource(jedis, isBroken);  
    120.         }  
    121.     }  
    122.   
    123.     /** 
    124.      *  取得key的值 
    125.      *  
    126.      * @param key 
    127.      */  
    128.     public static String stringGet(String key) {  
    129.         Jedis jedis = null;  
    130.         boolean isBroken = false;  
    131.         String lastVal = null;  
    132.         try {  
    133.             jedis = getJedis();  
    134.             jedis.select(0);  
    135.             lastVal = jedis.get(key);  
    136.             jedis.expire(key, expireTime);  
    137.         } catch (Exception e) {  
    138.             isBroken = true;  
    139.         } finally {  
    140.             closeResource(jedis, isBroken);  
    141.         }  
    142.         return lastVal;  
    143.     }  
    144.   
    145.     /** 
    146.      *  添加string数据 
    147.      *  
    148.      * @param key 
    149.      * @param value 
    150.      */  
    151.     public static String stringSet(String key, String value) {  
    152.         Jedis jedis = null;  
    153.         boolean isBroken = false;  
    154.         String lastVal = null;  
    155.         try {  
    156.             jedis = getJedis();  
    157.             jedis.select(0);  
    158.             lastVal = jedis.set(key, value);  
    159.             jedis.expire(key, expireTime);  
    160.         } catch (Exception e) {  
    161.             e.printStackTrace();  
    162.             isBroken = true;  
    163.         } finally {  
    164.             closeResource(jedis, isBroken);  
    165.         }  
    166.         return lastVal;  
    167.     }  
    168.   
    169.     /** 
    170.      *  添加hash数据 
    171.      *  
    172.      * @param key 
    173.      * @param field 
    174.      * @param value 
    175.      */  
    176.     public static void hashSet(String key, String field, String value) {  
    177.         boolean isBroken = false;  
    178.         Jedis jedis = null;  
    179.         try {  
    180.             jedis = getJedis();  
    181.             if (jedis != null) {  
    182.                 jedis.select(0);  
    183.                 jedis.hset(key, field, value);  
    184.                 jedis.expire(key, expireTime);  
    185.             }  
    186.         } catch (Exception e) {  
    187.             isBroken = true;  
    188.         } finally {  
    189.             closeResource(jedis, isBroken);  
    190.         }  
    191.     }  
    192.   
    193. }  

    注意：

    1，客户端的Jedis连接不同于项目里的Jedis连接需要Spring注解，直接使用静态方法就可以。

    运行
    1，运行canal服务端startup.bat / startup.sh
    2，运行客户端程序

    注意
    1，虽然canal服务端解析binlog后不会把数据持久化，但canal服务端会记录每次客户端消费的位置(客户端每次ack时服务端会记录pos点)。如果数据正在更新时，canal服务端挂掉，客户端也会跟着挂掉，mysql依然在插入数据，而redis则因为客户端的关闭而停止更新，造成mysql和redis的数据不一致。解决办法是，只要重启canal服务端和客户端就可以了，虽然canal服务端因为重启之前解析数据清空，但因为canal服务端记录的是客户端最后一次获取的pos点，canal服务端再从这个pos点开始解析，客户端更新至redis，以达到数据的一致。
    2，如果只有一个canal服务端和一个客户端，肯定存在可用性低的问题，一种做法是用程序来监控canal服务端和客户端，如果挂掉，再重启；一种做法是多个canal服务端+zk，将canal服务端的配置文件放在zk，任何一个canal服务端挂掉后，切换到其他canal服务端，读到的配置文件的内容就是一致的（还有记录的消费pos点），保证业务的高可用，客户端可使用相同的做法。

    转载 http://blog.csdn.net/stubborn_cow/article/details/50371405
  • 相关阅读:
    负载均衡session会话保持方法
    PHP分布式中Redis实现Session
    Nginx内置变量
    Nginx配置文件解析
    Nginx重写
    Nginx与Apache比较
    CGI概念
    Linux笔记（十四）
    Linux笔记（十三）
    hdu 4039
  • 原文地址：https://www.cnblogs.com/yaoyangding/p/12109400.html
Copyright © 2011-2022 走看看