网钛远程桌面管理助手
当前位置:首页 > 服务器资讯

服务器资讯

缓存一致性和跨服务器查询的数据异构解决方案

时间:2020-11-06 08:45:45   作者:   来源:   阅读:143   评论:0
内容摘要: 即将开播:4月29日,民生银行郭庆谈商业银行金融科技赋能的探索与实践--> 当项目的请求量上去了之后,通常有两种做法来应对高并发,第一是尽最大可能的使用cache来对抗,第二是尽最大可能的分库分表对抗。。。说起来容易,做起来并不那么乐观,这一篇就来浅析下。一:如何保证缓存......

即将开播:4月29日,民生银行郭庆谈商业银行金融科技赋能的探索与实践

-->

 

当项目的请求量上去了之后,通常有两种做法来应对高并发,第一是尽最大可能的使用cache来对抗,第二是尽最大可能的分库分表对抗。。。说起来容易,做起来并不那么乐观,这一篇就来浅析下。

一:如何保证缓存一致性

如我们的千人千面系统中,会针对商品,订单等多维度为某一个商家店铺自动化建立大约400个数据模型,然后买家在淘宝下订单之后,淘宝会将订单推送过来,订单会在400个模型中兜一圈,从而推送更贴切符合该买家行为习惯的触达,为了应对高并发,这些模型自然都是缓存在Cache中,如果有新的模型进来了,我如何通知redis进行缓存更新呢?通常的做法就是在添加模型的时候,顺便更新redis。。。对吧,如下图:

说的简单,web开发的程序员会说,麻蛋的,我管你什么业务,凭啥要我做推送,把我代码搞出问题了,你负责呀???所以你必须得碰一鼻子灰。就算搞定了web程序员,你可能还会遇到更新database成功,更新redis的时候失败,可人家不管,而且错误日志还在别人的日志系统里面,所以你很难甚至无法保证这个db和cache的缓存一致性,那这个时候能不能换个思路,我直接写个程序订阅database的binlog,从binlog中分析出模型数据的CURD操作,根据这些CURD的实际情况更新Redis的缓存数据,第一个可以实现和web的解耦,第二个实现了高度的缓存一致性,所以新的架构是这样的。

上面这张图,相信大家都能看得懂,重点就是这个处理binlog程序,从binlog中分析出CURD从而更新Redis,其实这个binlog程序就是本篇所说的canal。。。一个伪装成mysql的slave,不断的通过dump命令从mysql中盗出binlog日志,从而完美的实现了这个需求。

二:如何实现跨服务器 join 查询

本篇开头也说到了,数据量大了之后,必然会存在分库分表,甚至database都要分散到多台服务器上,现在的电商项目,都是业务赶着技术跑。。。谁也不知道下一个业务会是一个怎样的奇葩,所以必然会导致你要做一些跨服务器join查询,你以为自己很聪明,其实DBA早就把跨服务器查询的函数给你关掉了,求爹爹拜奶奶都不会给你开的,除非你杀一个DBA祭天,不过如果你的业务真的很重要,可能DBA会给你做数据异构,所谓的数据异构,那就是将需要join查询的多表按照某一个维度又聚合在一个DB中。让你去查询。。。。。

那如果用canal来订阅binlog,就可以改造成下面这种架构。

三:搭建一览

好了,canal的应用场景给大家也介绍到了,最主要是理解这种思想,人家搞不定的东西,你的价值就出来了。

1. 开启mysql的binlog功能

开启binlog,并且将binlog的格式改为Row,这样就可以获取到CURD的二进制内容,windows上的路径为:C:\ProgramData\MySQL\MySQL Server 5.7\my.ini。

  1. log-bin=mysql-bin #添加这一行就ok 
  2. binlog-format=ROW #选择row模式 
  3. server_id=1 

2. 验证binlog是否开启

使用命令验证,并且开启binlog的过期时间为30天,默认情况下binlog是不过期的,这就导致你的磁盘可能会爆满,直到挂掉。

  1. show variables like 'log_%';  
  2. #设置binlog的过期时间为30天 
  3. show variables like '%expire_logs_days%'
  4. set global expire_logs_days=30; 

3. 给canal服务器分配一个mysql的账号权限,方便canal去偷binlog日志。

  1. CREATE USER canal IDENTIFIED BY 'canal';     
  2. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';   
  3. FLUSH PRIVILEGES;   
  4.  
  5. show grants for 'canal' 

4. 下载canal

github的地址:https://github.com/alibaba/canal/releases

5. 然后就是各种tar解压 canal.deployer-1.0.24.tar.gz => canal

  1. [root@localhost myapp]# ls 
  2. apache-maven-3.5.0-bin.tar.gz                        dubbo-monitor-simple-2.5.4-SNAPSHOT.jar     nginx                tengine-2.2.0.tar.gz 
  3. canal                                                gearmand                                    nginx-1.13.4.tar.gz  tengine_st 
  4. canal.deployer-1.0.24.tar.gz                         gearmand-1.1.17                             nginx_st             tomcat 
  5. dubbo                                                gearmand-1.1.17.tar.gz                      redis                zookeeper 
  6. dubbo-monitor-simple-2.5.4-SNAPSHOT                  maven                                       redis-4.0.1.tar.gz   zookeeper-3.4.9.tar.gz 
  7. dubbo-monitor-simple-2.5.4-SNAPSHOT-assembly.tar.gz  mysql-5.7.19-linux-glibc2.12-x86_64.tar.gz  tengine 
  8. [root@localhost myapp]# cd canal 
  9. [root@localhost canal]# ls 
  10. bin  conf  lib  logs 
  11. [root@localhost canal]# cd conf 
  12. [root@localhost conf]# ls 
  13. canal.properties  example  logback.xml  spring 
  14. [root@localhost conf]# cd example 
  15. [root@localhost example]# ls 
  16. instance.properties  meta.dat 
  17. [root@localhost example]# 

6. canal 和 instance 配置文件

canal的模式是这样的,一个canal里面可能会有多个instance,也就说一个instance可以监控一个mysql实例,多个instance也就可以对应多台服务器的mysql实例。也就是一个canal就可以监控分库分表下的多机器mysql。

1) canal.properties

它是全局性的canal服务器配置,具体如下,这里面的参数涉及到方方面面。

  1. ################################################# 
  2. #########               common argument         #############  
  3. ################################################# 
  4. canal.id= 1 
  5. canal.ip= 
  6. canal.port= 11111 
  7. canal.zkServers= 
  8. # flush data to zk 
  9. canal.zookeeper.flush.period = 1000 
  10. # flush meta cursor/parse position to file 
  11. canal.file.data.dir = ${canal.conf.dir} 
  12. canal.file.flush.period = 1000 
  13. ## memory store RingBuffer size, should be Math.pow(2,n) 
  14. canal.instance.memory.buffer.size = 16384 
  15. ## memory store RingBuffer used memory unit size , default 1kb 
  16. canal.instance.memory.buffer.memunit = 1024  
  17. ## meory store gets mode used MEMSIZE or ITEMSIZE 
  18. canal.instance.memory.batch.mode = MEMSIZE 
  19.      
  20. ## detecing config 
  21. canal.instance.detecting.enable = false 
  22. #canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now() 
  23. canal.instance.detecting.sql = select 1 
  24. canal.instance.detecting.interval.time = 3 
  25. canal.instance.detecting.retry.threshold = 3 
  26. canal.instance.detecting.heartbeatHaEnable = false 
  27.  
  28. # support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery 
  29. canal.instance.transaction.size =  1024 
  30. # mysql fallback connected to new master should fallback times 
  31. canal.instance.fallbackIntervalInSeconds = 60 
  32.  
  33. # network config 
  34. canal.instance.network.receiveBufferSize = 16384 
  35. canal.instance.network.sendBufferSize = 16384 
  36. canal.instance.network.soTimeout = 30 
  37.  
  38. # binlog filter config 
  39. canal.instance.filter.query.dcl = false 
  40. canal.instance.filter.query.dml = false 
  41. canal.instance.filter.query.ddl = false 
  42. canal.instance.filter.table.error = false 
  43. canal.instance.filter.rows = false 
  44.  
  45. # binlog format/image check 
  46. canal.instance.binlog.format = ROW,STATEMENT,MIXED  
  47. canal.instance.binlog.image = FULL,MINIMAL,NOBLOB 
  48.  
  49. # binlog ddl isolation 
  50. canal.instance.get.ddl.isolation = false 
  51.  
  52. ################################################# 
  53. #########               destinations            #############  
  54. ################################################# 
  55. canal.destinations= example 
  56. # conf root dir 
  57. canal.conf.dir = ../conf 
  58. # auto scan instance dir add/remove and start/stop instance 
  59. canal.auto.scan = true 
  60. canal.auto.scan.interval = 5 
  61.  
  62. canal.instance.global.mode = spring  
  63. canal.instance.global.lazy = false 
  64. #canal.instance.global.manager.address = 127.0.0.1:1099 
  65. #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml 
  66. canal.instance.global.spring.xml = classpath:spring/file-instance.xml 
  67. #canal.instance.global.spring.xml = classpath:spring/default-instance.xml 
  68.  
  69. #################################################   
  70. ## mysql serverId   
  71. canal.instance.mysql.slaveId = 1234   
  72.  
  73. # position info,需要改成自己的数据库信息   
  74. canal.instance.master.address = 127.0.0.1:3306    
  75. canal.instance.master.journal.name = 
  76. canal.instance.master.position = 
  77. canal.instance.master.timestamp = 
  78.  
  79. #canal.instance.standby.address =    
  80. #canal.instance.standby.journal.name =   
  81. #canal.instance.standby.position =    
  82. #canal.instance.standby.timestamp =    
  83.  
  84. # username/password,需要改成自己的数据库信息   
  85. canal.instance.dbUsername = root 
  86. canal.instance.dbPassword = 123456 
  87. canal.instance.defaultDatabaseName = datamip   
  88. canal.instance.connectionCharset = UTF-8   
  89.  
  90. table regex   
  91. canal.instance.filter.regex = .*\\..*   
  92.  
  93. ################################################# 

由于是全局性的配置,所以上面三处标红的地方要注意一下:

  • canal.port= 11111 当前canal的服务器端口号
  • canal.destinations= example 当前默认开启了一个名为example的instance实例,如果想开多个instance,用","逗号隔开就可以了。。。
  • canal.instance.filter.regex = .\.. mysql实例下的所有db的所有表都在监控范围内。

2) instance.properties

这个就是具体的某个instances实例的配置,未涉及到的配置都会从canal.properties上继承。

  1. ################################################# 
  2. ## mysql serverId 
  3. canal.instance.mysql.slaveId = 1234 
  4.  
  5. # position info 
  6. canal.instance.master.address = 192.168.23.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 =datamip 
  20. canal.instance.connectionCharset = UTF-8 
  21.  
  22. table regex 
  23. canal.instance.filter.regex = .*\\..* 
  24. table black regex 
  25. canal.instance.filter.black.regex = 
  26.  
  27. ################################################# 

上面标红的地方注意下就好了,去偷binlog的时候,需要知道的mysql地址和用户名,密码。

7. 开启canal

大家要记得把 /canal/bin 目录配置到 /etc/profile 的 Path中,方便快速开启,通过下图你会看到11111端口已经在centos上开启了。

  1. [root@localhost bin]# ls 
  2. canal.pid  startup.bat  startup.sh  stop.sh 
  3. [root@localhost bin]# pwd 
  4. /usr/myapp/canal/bin 
  5. [root@localhost example]# startup.sh 
  6. cd to /usr/myapp/canal/bin for workaround relative path 
  7. LOG CONFIGURATION : /usr/myapp/canal/bin/../conf/logback.xml 
  8. canal conf : /usr/myapp/canal/bin/../conf/canal.properties 
  9. CLASSPATH :/usr/myapp/canal/bin/../conf:/usr/myapp/canal/bin/../lib/zookeeper-3.4.5.jar:/usr/myapp/canal/bin/../lib/zkclient-0.1.jar:/usr/myapp/canal/bin/../lib/spring-2.5.6.jar:/usr/myapp/canal/bin/../lib/slf4j-api-1.7.12.jar:/usr/myapp/canal/bin/../lib/protobuf-java-2.6.1.jar:/usr/myapp/canal/bin/../lib/oro-2.0.8.jar:/usr/myapp/canal/bin/../lib/netty-all-4.1.6.Final.jar:/usr/myapp/canal/bin/../lib/netty-3.2.5.Final.jar:/usr/myapp/canal/bin/../lib/logback-core-1.1.3.jar:/usr/myapp/canal/bin/../lib/logback-classic-1.1.3.jar:/usr/myapp/canal/bin/../lib/log4j-1.2.14.jar:/usr/myapp/canal/bin/../lib/jcl-over-slf4j-1.7.12.jar:/usr/myapp/canal/bin/../lib/guava-18.0.jar:/usr/myapp/canal/bin/../lib/fastjson-1.2.28.jar:/usr/myapp/canal/bin/../lib/commons-logging-1.1.1.jar:/usr/myapp/canal/bin/../lib/commons-lang-2.6.jar:/usr/myapp/canal/bin/../lib/commons-io-2.4.jar:/usr/myapp/canal/bin/../lib/commons-beanutils-1.8.2.jar:/usr/myapp/canal/bin/../lib/canal.store-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.sink-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.server-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.protocol-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.parse.driver-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.parse.dbsync-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.parse-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.meta-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.instance.spring-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.instance.manager-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.instance.core-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.filter-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.deployer-1.0.24.jar:/usr/myapp/canal/bin/../lib/canal.common-1.0.24.jar:/usr/myapp/canal/bin/../lib/aviator-2.2.1.jar: 
  10. cd to /usr/myapp/canal/conf/example for continue 
  11. [root@localhost example]# netstat -tln 
  12. Active Internet connections (only servers) 
  13. Proto Recv-Q Send-Q Local Address           Foreign Address         State       
  14. tcp        0      0 0.0.0.0:11111           0.0.0.0:*               LISTEN      
  15. tcp        0      0 0.0.0.0:111             0.0.0.0:*               LISTEN      
  16. tcp        0      0 192.168.122.1:53        0.0.0.0:*               LISTEN      
  17. tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN      
  18. tcp        0      0 127.0.0.1:631           0.0.0.0:*               LISTEN      
  19. tcp        0      0 127.0.0.1:25            0.0.0.0:*               LISTEN      
  20. tcp6       0      0 :::111                  :::*                    LISTEN      
  21. tcp6       0      0 :::22                   :::*                    LISTEN      
  22. tcp6       0      0 ::1:631                 :::*                    LISTEN      
  23. tcp6       0      0 ::1:25                  :::*                    LISTEN      
  24. [root@localhost example]# 

8. Java Client 代码

canal driver 需要在maven仓库中获取一下:http://www.mvnrepository.com/artifact/com.alibaba.otter/canal.client/1.0.24,不过依赖还是蛮多的。

  1. <!-- https://mvnrepository.com/artifact/com.alibaba.otter/canal.client --> 
  2.   <dependency> 
  3.       <groupId>com.alibaba.otter</groupId> 
  4.       <artifactId>canal.client</artifactId> 
  5.       <version>1.0.24</version> 
  6.   </dependency> 

9. 启动java代码进行验证

下面的代码对table的CURD都做了一个基本的判断,看看是不是能够智能感知,然后可以根据实际情况进行redis的更新操作。。。

  1. package com.datamip.canal; 
  2.  
  3. import java.awt.Event; 
  4. import java.net.InetSocketAddress; 
  5. import java.util.List; 
  6.  
  7. import com.alibaba.otter.canal.client.CanalConnector; 
  8. import com.alibaba.otter.canal.client.CanalConnectors; 
  9. import com.alibaba.otter.canal.protocol.CanalEntry.Column
  10. import com.alibaba.otter.canal.protocol.CanalEntry.Entry; 
  11. import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; 
  12. import com.alibaba.otter.canal.protocol.CanalEntry.EventType; 
  13. import com.alibaba.otter.canal.protocol.CanalEntry.Header; 
  14. import com.alibaba.otter.canal.protocol.CanalEntry.RowChange; 
  15. import com.alibaba.otter.canal.protocol.Message; 
  16. import com.google.protobuf.InvalidProtocolBufferException; 
  17.  
  18. public class App { 
  19.  
  20.     public static void main(String[] args) throws InterruptedException { 
  21.  
  22.         // 第一步:与canal进行连接 
  23.         CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.23.170", 11111), 
  24.                 "example"""""); 
  25.         connector.connect(); 
  26.  
  27.         // 第二步:开启订阅 
  28.         connector.subscribe(); 
  29.  
  30.         // 第三步:循环订阅 
  31.         while (true) { 
  32.             try { 
  33.                 // 每次读取 1000 条 
  34.                 Message message = connector.getWithoutAck(1000); 
  35.  
  36.                 long batchID = message.getId(); 
  37.  
  38.                 int size = message.getEntries().size(); 
  39.  
  40.                 if (batchID == -1 || size == 0) { 
  41.                     System.out.println("当前暂时没有数据"); 
  42.                     Thread.sleep(1000); // 没有数据 
  43.                 } else { 
  44.                     System.out.println("-------------------------- 有数据啦 -----------------------"); 
  45.                     PrintEntry(message.getEntries()); 
  46.                 } 
  47.  
  48.                 // position id ack (方便处理下一条) 
  49.                 connector.ack(batchID); 
  50.  
  51.             } catch (Exception e) { 
  52.                 // TODO: handle exception 
  53.  
  54.             } finally { 
  55.                 Thread.sleep(1000); 
  56.             } 
  57.         } 
  58.     } 
  59.  
  60.     // 获取每条打印的记录 
  61.     @SuppressWarnings("static-access"
  62.     public static void PrintEntry(List<Entry> entrys) { 
  63.  
  64.         for (Entry entry : entrys) { 
  65.  
  66.             // 第一步:拆解entry 实体 
  67.             Header header = entry.getHeader(); 
  68.             EntryType entryType = entry.getEntryType(); 
  69.  
  70.             // 第二步: 如果当前是RowData,那就是我需要的数据 
  71.             if (entryType == EntryType.ROWDATA) { 
  72.  
  73.                 String tableName = header.getTableName(); 
  74.                 String schemaName = header.getSchemaName(); 
  75.  
  76.                 RowChange rowChange = null
  77.  
  78.                 try { 
  79.                     rowChange = RowChange.parseFrom(entry.getStoreValue()); 
  80.                 } catch (InvalidProtocolBufferException e) { 
  81.                     e.printStackTrace(); 
  82.                 } 
  83.  
  84.                 EventType eventType = rowChange.getEventType(); 
  85.  
  86.                 System.out.println(String.format("当前正在操作 %s.%s, Action= %s", schemaName, tableName, eventType)); 
  87.  
  88.                 // 如果是‘查询’ 或者 是 ‘DDL’ 操作,那么sql直接打出来 
  89.                 if (eventType == EventType.QUERY || rowChange.getIsDdl()) { 
  90.                     System.out.println("rowchange sql ----->" + rowChange.getSql()); 
  91.                     return
  92.                 } 
  93.  
  94.                 // 第三步:追踪到 columns 级别 
  95.                 rowChange.getRowDatasList().forEach((rowData) -> { 
  96.  
  97.                     // 获取更新之前的column情况 
  98.                     List<Column> beforeColumns = rowData.getBeforeColumnsList(); 
  99.  
  100.                     // 获取更新之后的 column 情况 
  101.                     List<Column> afterColumns = rowData.getAfterColumnsList(); 
  102.  
  103.                     // 当前执行的是 删除操作 
  104.                     if (eventType == EventType.DELETE) { 
  105.                         PrintColumn(beforeColumns); 
  106.                     } 
  107.  
  108.                     // 当前执行的是 插入操作 
  109.                     if (eventType == eventType.INSERT) { 
  110.                         PrintColumn(afterColumns); 
  111.                     } 
  112.  
  113.                     // 当前执行的是 更新操作 
  114.                     if (eventType == eventType.UPDATE) { 
  115.                         PrintColumn(afterColumns); 
  116.                     } 
  117.                 }); 
  118.             } 
  119.         } 
  120.     } 
  121.  
  122.     // 每个row上面的每一个column 的更改情况 
  123.     public static void PrintColumn(List<Column> columns) { 
  124.  
  125.         columns.forEach((column) -> { 
  126.  
  127.             String columnName = column.getName(); 
  128.             String columnValue = column.getValue(); 
  129.             String columnType = column.getMysqlType(); 
  130.             boolean isUpdated = column.getUpdated(); // 判断 该字段是否更新 
  131.  
  132.             System.out.println(String.format("columnName=%s, columnValue=%s, columnType=%s, isUpdated=%s", columnName, 
  133.                     columnValue, columnType, isUpdated)); 
  134.  
  135.         }); 
  136.  
  137.     } 

Update操作

Insert操作

Delete 操作

从结果中看,没毛病,有图有真相,好了,本篇就说到这里,对于开发的你,肯定是有帮助的~~~

本文转载自微信公众号「 一线码农聊技术」,可以通过以下二维码关注。转载本文请联系 一线码农聊技术公众号。


标签:canal  instance  myapp  bin  usr  

相关文章

相关评论