使用canal进行mysql数据同步到Redis

本文涉及的产品
云数据库 Tair(兼容Redis),内存型 2GB
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
简介:

1. 可行方案

回归正题:我们的标题为《使用canal进行MySQL数据同步到Redis》,那就先来说说我们的目的:mysql数据同步到Redis,想达到读写分离,Redis只做缓存,MySQL做持久化。刚开始想这样干的时候就去网上收集资料,发现了N多做法:

  1. 先从Redis读取数据,如果没有查询到;便从mysql查询数据,将查询到的内容放到Redis中。对于写操作,先对mysql进行写,写成功对Redis进行写。当然这是一种相对直观而且简单的方法,但是看起来有许多操作需要我们自己去做。

  2. 使用mysql的udf去做,大体的思想是通过数据库中的Trigger调用自定义的函数库来触发对Redis的相应操作,比较麻烦的一点是:自定义的函数库需要我们基于mysql的API进行开发(C++),想想自己的Java程序要去调用这么一堆玩意,本人很不情愿。据了解,该方法也是阿里早起的解决方案,具体的步骤可参照:《【菜鸟玩Linux开发】通过MySQL自动同步刷新Redis》

  3. 通过Gearman去同步,但是通过了解发现,它一般使用在PHP的开发中。

    接下来的两种方案都属于对mysql中的binlog进行解析的方法了。

  4. 使用open-replicator解析binlog,https://github.com/whitesock/open-replicator.

  5. 使用canal进行同步,当然是能够解放双手的工具。

通过大量的资料收集和调查,我使用了canal进行了mysql数据同步到Redis。先简单谈谈canal:

canal主要是基于数据库的日志解析,获取增量变更进行同步,由此衍生出了增量订阅&消费的业务,核心基本就是模拟mysql中slave节点请求。具体的原理在这里不进行介绍,可以移步《阿里巴巴开源项目: canal 基于mysql数据库binlog的增量订阅&消费》 进行学习。

2. mysql的配置

  • 开启mysql的binlog模块

切换到mysql的安装路径,找到my.cnf(Linux)/my.ini (windows),加入如下内容:

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

配置完成后,需要重启数据库。当重启数据库遇到问题时,耐心解决,但需要警告的是,千万别动data文件夹下的文件。当然如果你觉得你比较有“资本”,同时遇到了“mysql 1067 无法启动”的错误,你可以试着备份一下data文件夹下的内容,删除logfile文件,重启数据库即可,但本人极不推荐这样进行操作。就是由于本人之前的无知,根据一个无良博客,误删了ibdata1文件,使得本人造成了很大的损失,mysql下的所有数据库瞬间毁灭。

  • 配置mysql数据库

创建canal用户,用来管理canal的访问权限。我们可以通过对canal用户访问权限的控制,进而控制canal能够获取的内容。

1
2
CREATE USER canal IDENTIFIED BY  'canal' ;  
   GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON 数据库名.表名 TO  'canal' @ '%' ;  -- GRANT ALL PRIVILEGES ON 数据库名.表名 TO  'canal' @ '%'  ;  FLUSH PRIVILEGES;

3. canal配置与部署

  • 下载部署包

下载,解压,我使用的是最新版本1.0.22

1
https: //github.com/alibaba/canal/releases/
  • 配置canal

主要配置的文件有两处,canal/conf/example/instance.properties 和 canal/conf/canal.properties . 而canal.properties 文件我们一般保持默认配置,所以我们仅对instance.properties 进行修改。如果需要对canal进行复杂的配置,可以参考《Canal AdminGuide》

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
## mysql serverId
canal.instance.mysql.slaveId =  1234
 
# position info
canal.instance.master.address = ***.***.***.***: 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 = .*\\..*
# table black regex
canal.instance.filter.black.regex =
  • 启动canal

./canal/startup.sh
  • 查看启动状态

我们可以通过查看logs/canal/canal.log 和logs/example/example.log日志来判断canal是否启动成功。

canal/logs/canal/canal.log

1
2
3
2016 - 12 - 29  14 : 03 : 00.956  [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
2016 - 12 - 29  14 : 03 : 01.071  [main] INFO  com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[ 192.168 . 1.99 : 11111 ]
2016 - 12 - 29  14 : 03 : 01.628  [main] INFO  com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ....


canal/logs/example/example.log

1
2
3
4
2016 - 12 - 29  14 : 03 : 01.357  [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from  class  path resource [canal.properties]
2016 - 12 - 29  14 : 03 : 01.362  [main] INFO  c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from  class  path resource [example/instance.properties]
2016 - 12 - 29  14 : 03 : 01.535  [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance  for  1 -example 
2016 - 12 - 29  14 : 03 : 01.555  [main] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....

4. Java连接canal执行同步操作

在maven项目中中加载canal和redis依赖包.

1
2
3
4
5
6
7
8
9
10
dependency>    
     <groupId>redis.clients</groupId>    
     <artifactId>jedis</artifactId>    
     <version> 2.4 . 2 </version>    
</dependency> 
<dependency>    
     <groupId>com.alibaba.otter</groupId>    
     <artifactId>canal.client</artifactId>    
     <version> 1.0 . 22 </version>    
</dependency>

建立canal客户端,从canal中获取数据,并将数据更新至Redis.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
import  java.net.InetSocketAddress;  
import  java.util.List;  
 
import  com.alibaba.fastjson.JSONObject;
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;  
import  com.alibaba.otter.canal.client.*;  
 
public  class  CanalClient{  
    public  static  void  main(String args[]) {  
        CanalConnector connector = CanalConnectors.newSingleConnector( new  InetSocketAddress(AddressUtils.getHostIp(),  
                11111 ),  "example" "" "" );  
        int  batchSize =  1000 ;  
        try  {  
            connector.connect();  
            connector.subscribe( ".*\\..*" );  
            connector.rollback();    
            while  ( true ) {  
                Message message = connector.getWithoutAck(batchSize);  // 获取指定数量的数据  
                long  batchId = message.getId();  
                int  size = message.getEntries().size();  
                if  (batchId == - 1  || size ==  0 ) {  
                    try  {  
                        Thread.sleep( 1000 );  
                    catch  (InterruptedException e) {  
                        e.printStackTrace();  
                    }  
                else  {  
                    printEntry(message.getEntries());  
                }  
                connector.ack(batchId);  // 提交确认  
                // connector.rollback(batchId); // 处理失败, 回滚数据  
            }  
        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( "================> 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) {  
                    redisDelete(rowData.getBeforeColumnsList());  
                else  if  (eventType == EventType.INSERT) {  
                    redisInsert(rowData.getAfterColumnsList());  
                else  {  
                    System.out.println( "-------> before" );  
                    printColumn(rowData.getBeforeColumnsList());  
                    System.out.println( "-------> after" );  
                    redisUpdate(rowData.getAfterColumnsList());  
                }  
            }  
        }  
    }  
 
    private  static  void  printColumn( List<Column> columns) {  
        for  (Column column : columns) {  
            System.out.println(column.getName() +  " : "  + column.getValue() +  "    update="  + column.getUpdated());  
        }  
    }  
 
   private  static  void  redisInsert( List<Column> columns){
       JSONObject json= new  JSONObject();
       for  (Column column : columns) {  
           json.put(column.getName(), column.getValue());  
        }  
       if (columns.size()> 0 ){
           RedisUtil.stringSet( "user:" + columns.get( 0 ).getValue(),json.toJSONString());
       }
    }
 
   private  static   void  redisUpdate( List<Column> columns){
       JSONObject json= new  JSONObject();
       for  (Column column : columns) {  
           json.put(column.getName(), column.getValue());  
        }  
       if (columns.size()> 0 ){
           RedisUtil.stringSet( "user:" + columns.get( 0 ).getValue(),json.toJSONString());
       }
   }
 
    private  static   void  redisDelete( List<Column> columns){
        JSONObject json= new  JSONObject();
           for  (Column column : columns) {  
               json.put(column.getName(), column.getValue());  
            }  
           if (columns.size()> 0 ){
               RedisUtil.delKey( "user:" + columns.get( 0 ).getValue());
           }
    }   
}  
 
RedisUtil 工具类
 
import  redis.clients.jedis.Jedis;
import  redis.clients.jedis.JedisPool;
import  redis.clients.jedis.JedisPoolConfig;
 
public  class  RedisUtil {
 
     // Redis服务器IP
     private  static  String ADDR =  "0.0.0.0" ;
 
     // Redis的端口号
     private  static  int  PORT =  6379 ;
 
     // 访问密码
     //private static String AUTH = "admin";
 
     // 可用连接实例的最大数目,默认值为8;
     // 如果赋值为-1,则表示不限制;如果pool已经分配了maxActive个jedis实例,则此时pool的状态为exhausted(耗尽)。
     private  static  int  MAX_ACTIVE =  1024 ;
 
     // 控制一个pool最多有多少个状态为idle(空闲的)的jedis实例,默认值也是8。
     private  static  int  MAX_IDLE =  200 ;
 
     // 等待可用连接的最大时间,单位毫秒,默认值为-1,表示永不超时。如果超过等待时间,则直接抛出JedisConnectionException;
     private  static  int  MAX_WAIT =  10000 ;
 
     // 过期时间
     protected  static  int   expireTime =  60  60  * 24 ;
 
     // 连接池
     protected  static  JedisPool pool;
 
     /**
      * 静态代码,只在初次调用一次
      */
     static  {
         JedisPoolConfig config =  new  JedisPoolConfig();
         //最大连接数
         config.setMaxTotal(MAX_ACTIVE);
         //最多空闲实例
         config.setMaxIdle(MAX_IDLE);
         //超时时间
         config.setMaxWaitMillis(MAX_WAIT);
         //
         config.setTestOnBorrow( false );
         pool =  new  JedisPool(config, ADDR, PORT,  1000 );
     }
 
     /**
      * 获取jedis实例
      */
     protected  static  synchronized  Jedis getJedis() {
         Jedis jedis =  null ;
         try  {
             jedis = pool.getResource();
         catch  (Exception e) {
             e.printStackTrace();
             if  (jedis !=  null ) {
                 pool.returnBrokenResource(jedis);
             }
         }
         return  jedis;
     }
 
     /**
      * 释放jedis资源
      * @param jedis
      * @param isBroken
      */
     protected  static  void  closeResource(Jedis jedis,  boolean  isBroken) {
         try  {
             if  (isBroken) {
                 pool.returnBrokenResource(jedis);
             else  {
                 pool.returnResource(jedis);
             }
         catch  (Exception e) {
 
         }
     }
 
     /**
      * 是否存在key
      * @param key
      */
     public  static  boolean  existKey(String key) {
         Jedis jedis =  null ;
         boolean  isBroken =  false ;
         try  {
             jedis = getJedis();
             jedis.select( 0 );
             return  jedis.exists(key);
         catch  (Exception e) {
             isBroken =  true ;
         finally  {
             closeResource(jedis, isBroken);
         }
         return  false ;
     }
 
     /**
      * 删除key
      * @param key
      */
     public  static  void  delKey(String key) {
         Jedis jedis =  null ;
         boolean  isBroken =  false ;
         try  {
             jedis = getJedis();
             jedis.select( 0 );
             jedis.del(key);
         catch  (Exception e) {
             isBroken =  true ;
         finally  {
             closeResource(jedis, isBroken);
         }
     }
 
     /**
      * 取得key的值
      * @param key
      */
     public  static  String stringGet(String key) {
         Jedis jedis =  null ;
         boolean  isBroken =  false ;
         String lastVal =  null ;
         try  {
             jedis = getJedis();
             jedis.select( 0 );
             lastVal = jedis.get(key);
             jedis.expire(key, expireTime);
         catch  (Exception e) {
             isBroken =  true ;
         finally  {
             closeResource(jedis, isBroken);
         }
         return  lastVal;
     }
 
     /**
      * 添加string数据
      * @param key
      * @param value
      */
     public  static  String stringSet(String key, String value) {
         Jedis jedis =  null ;
         boolean  isBroken =  false ;
         String lastVal =  null ;
         try  {
             jedis = getJedis();
             jedis.select( 0 );
             lastVal = jedis.set(key, value);
             jedis.expire(key, expireTime);
         catch  (Exception e) {
             e.printStackTrace();
             isBroken =  true ;
         finally  {
             closeResource(jedis, isBroken);
         }
         return  lastVal;
     }
 
     /**
      *  添加hash数据
      * @param key
      * @param field
      * @param value
      */
     public  static  void  hashSet(String key, String field, String value) {
         boolean  isBroken =  false ;
         Jedis jedis =  null ;
         try  {
             jedis = getJedis();
             if  (jedis !=  null ) {
                 jedis.select( 0 );
                 jedis.hset(key, field, value);
                 jedis.expire(key, expireTime);
             }
         catch  (Exception e) {
             isBroken =  true ;
         finally  {
             closeResource(jedis, isBroken);
         }
     }
}

至此,我们利用canal进行了mysql数据同步到Redis的任务,可以根据不同的需求将代码进行修改置于需要的位置。

参考:

 z转载:http://blog.csdn.net/tb3039450/article/details/53928351













本文转自yunlielai51CTO博客,原文链接:http://blog.51cto.com/4925054/1910483,如需转载请自行联系原作者

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
11天前
|
缓存 NoSQL 关系型数据库
Redis与MySQL的数据一致性
在高并发环境下,保持 Redis 和 MySQL 的数据一致性是一个复杂但重要的问题。通过采用读写穿透、写穿透、分布式锁、双写一致性保障和延时双删策略,可以有效地减少数据不一致的风险,确保系统的稳定性和可靠性。通过合理的缓存策略和数据同步机制,可以显著提升系统的性能和用户体验。
71 22
|
2月前
|
NoSQL 关系型数据库 Redis
《docker高级篇(大厂进阶):1.Docker复杂安装详说》包括:安装mysql主从复制、安装redis集群
《docker高级篇(大厂进阶):1.Docker复杂安装详说》包括:安装mysql主从复制、安装redis集群
119 14
|
1月前
|
关系型数据库 MySQL 应用服务中间件
《docker基础篇:8.Docker常规安装简介》包括:docker常规安装总体步骤、安装tomcat、安装mysql、安装redis
《docker基础篇:8.Docker常规安装简介》包括:docker常规安装总体步骤、安装tomcat、安装mysql、安装redis
114 7
|
2月前
|
NoSQL Java 关系型数据库
Liunx部署java项目Tomcat、Redis、Mysql教程
本文详细介绍了如何在 Linux 服务器上安装和配置 Tomcat、MySQL 和 Redis,并部署 Java 项目。通过这些步骤,您可以搭建一个高效稳定的 Java 应用运行环境。希望本文能为您在实际操作中提供有价值的参考。
171 26
|
2月前
|
NoSQL 关系型数据库 MySQL
Linux安装jdk、mysql、redis
Linux安装jdk、mysql、redis
209 7
|
4天前
|
缓存 NoSQL 中间件
Redis,分布式缓存演化之路
本文介绍了基于Redis的分布式缓存演化,探讨了分布式锁和缓存一致性问题及其解决方案。首先分析了本地缓存和分布式缓存的区别与优劣,接着深入讲解了分布式远程缓存带来的并发、缓存失效(穿透、雪崩、击穿)等问题及应对策略。文章还详细描述了如何使用Redis实现分布式锁,确保高并发场景下的数据一致性和系统稳定性。最后,通过双写模式和失效模式讨论了缓存一致性问题,并提出了多种解决方案,如引入Canal中间件等。希望这些内容能为读者在设计分布式缓存系统时提供有价值的参考。感谢您的阅读!
Redis,分布式缓存演化之路
|
2月前
|
存储 缓存 NoSQL
解决Redis缓存数据类型丢失问题
解决Redis缓存数据类型丢失问题
186 85
|
4月前
|
消息中间件 缓存 NoSQL
Redis 是一个高性能的键值对存储系统,常用于缓存、消息队列和会话管理等场景。
【10月更文挑战第4天】Redis 是一个高性能的键值对存储系统,常用于缓存、消息队列和会话管理等场景。随着数据增长,有时需要将 Redis 数据导出以进行分析、备份或迁移。本文详细介绍几种导出方法:1)使用 Redis 命令与重定向;2)利用 Redis 的 RDB 和 AOF 持久化功能;3)借助第三方工具如 `redis-dump`。每种方法均附有示例代码,帮助你轻松完成数据导出任务。无论数据量大小,总有一款适合你。
99 6
|
28天前
|
存储 缓存 NoSQL
云端问道21期方案教学-应对高并发,利用云数据库 Tair(兼容 Redis®*)缓存实现极速响应
云端问道21期方案教学-应对高并发,利用云数据库 Tair(兼容 Redis®*)缓存实现极速响应
|
28天前
|
缓存 NoSQL 关系型数据库
云端问道21期实操教学-应对高并发,利用云数据库 Tair(兼容 Redis®)缓存实现极速响应
本文介绍了如何通过云端问道21期实操教学,利用云数据库 Tair(兼容 Redis®)缓存实现高并发场景下的极速响应。主要内容分为四部分:方案概览、部署准备、一键部署和完成及清理。方案概览中,展示了如何使用 Redis 提升业务性能,降低响应时间;部署准备介绍了账号注册与充值步骤;一键部署详细讲解了创建 ECS、RDS 和 Redis 实例的过程;最后,通过对比测试验证了 Redis 缓存的有效性,并指导用户清理资源以避免额外费用。