Shulou(Shulou.com)05/31 Report--

Today, I will talk to you about cache consistency and data heterogeneity solutions for cross-server querying. Many people may not know much about it. In order to make you understand better, the editor summarized the following. I hope you can get something from this article.

When the number of requests for a project goes up, there are usually two ways to deal with high concurrency. The first is to use cache to combat as much as possible, and the second is to fight against the database and table as much as possible. It's easy to say, but not so optimistic to do. Let's analyze it below.

One: how to ensure cache consistency

For example, in our thousand-person thousand-noodle system, we will automatically build about 400 data models for a store for goods, orders and other dimensions, and then after the buyer places an order on Taobao, Taobao will push the order, and the order will circle around the 400 models, so as to push the touch that is more in line with the buyer's behavior habits. In order to cope with high concurrency, these models are naturally cached in Cache. If a new model comes in, how can I notify redis for cache updates? The usual practice is to update redis by the way when adding the model. Right, as shown in the following picture:

To put it simply, the programmer of web development will say, Ma egg, what do I care about your business? why do you want me to push and make my code go wrong? are you responsible? So you have to be snuffed out. Even if you get rid of the web programmer, you may still encounter the success of updating database and the failure of updating redis, but they don't care, and the error log is still in someone else's log system, so it is difficult or even impossible to guarantee the cache consistency of this db and cache, so can you change your way of thinking at this time? I will write a program directly to subscribe to the binlog of database and analyze the CURD operation of the model data from binlog. Update the cache data of Redis according to the actual situation of these CURD, the first can be decoupled from web, and the second can achieve a high degree of cache consistency, so the new architecture is like this.

The above picture, I believe everyone can understand, the point is to deal with the binlog program, analyze the CURD from the binlog to update the Redis, in fact, this binlog program is the canal in this article. A slave disguised as mysql keeps stealing binlog logs from mysql through dump commands, thus realizing this requirement perfectly.

Second: how to implement cross-server join query

As mentioned at the beginning of this article, after a large amount of data, there will inevitably be sub-databases and tables, and even database will be scattered to multiple servers. No one knows what kind of strange business the next business will be, so it will inevitably cause you to do some cross-server join queries. You think you are very smart. In fact, DBA has already turned off the functions of cross-server queries for you. Asking your parents and grandparents will not open them for you unless you kill a DBA, but if your business is really important, DBA may do data heterogeneity for you, the so-called data heterogeneity. That is, multiple tables that need to be queried by join are aggregated into a DB according to a certain dimension. I want you to check it out.

Then if you use canal to subscribe to binlog, you can transform it into the following architecture.

Three: build a list

Well, the application scenario of canal has also been introduced to everyone, the most important thing is to understand this kind of thinking, what others can't handle, your value will come out.

1. Enable the binlog function of mysql

Turn on binlog and change the format of binlog to Row, so you can get the binary content of CURD. The path on windows is: C:\ ProgramData\ MySQL\ MySQL Server 5.7\ my.ini.

Log-bin=mysql-bin # add this line to ok binlog-format=ROW # Select row mode server_id=1

two。 Verify that binlog is turned on

Use command verification, and the expiration time for binlog is 30 days, and binlog is not expired by default, which causes your disk to be full until you hang up.

Show variables like 'log_%'; # sets the expiration time of binlog to 30 days show variables like'% expire_logs_days%'; set global expire_logs_days=30

3. Assign a mysql account to the canal server to make it easier for canal to steal binlog logs.

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

4. Download canal

Address of github: https://github.com/alibaba/canal/releases

5. Then there are all kinds of tar decompression canal.deployer-1.0.24.tar.gz = > canal

[root@localhost myapp] # ls apache-maven-3.5.0-bin.tar.gz dubbo-monitor-simple-2.5.4-SNAPSHOT.jar nginx tengine-2.2.0.tar.gz canal gearmand nginx-1.13.4. Tar.gz tengine_st canal.deployer-1.0.24.tar.gz gearmand-1.1.17 nginx_st tomcat dubbo gearmand-1.1.17.tar.gz redis zookeeper dubbo-monitor- Simple-2.5.4-SNAPSHOT maven redis-4.0.1.tar.gz zookeeper-3.4.9.tar.gz dubbo-monitor-simple-2.5.4-SNAPSHOT-assembly.tar.gz mysql-5.7.19-linux-glibc2.12-x86_64.tar.gz tengine [root@localhost myapp] # cd canal [root@localhost canal] # ls bin conf Lib logs [root@localhost canal] # cd conf [root@localhost conf] # ls canal.properties example logback.xml spring [root@localhost conf] # cd example [root@localhost example] # ls instance.properties meta.dat [root@localhost example] #

6. Canal and instance configuration files

The mode of canal is that there may be multiple instance in a canal, that is, an instance can monitor a mysql instance, and multiple instance can correspond to mysql instances of multiple servers. That is, a canal can monitor the multi-machine mysql under the sub-database sub-table.

1) canal.properties

It is a global canal server configuration, as follows, in which the parameters are related to all aspects.

# common argument # # canal.id= 1 canal.ip= canal.port= 11111 canal.zkServers= # flush data to zk canal.zookeeper.flush.period = 1000 # flush meta cursor/parse position to file canal.file.data.dir = ${canal.conf.dir} canal.file.flush.period = 1000 # # memory store RingBuffer size Should be Math.pow (2) canal.instance.memory.buffer.size = 16384 # # memory store RingBuffer used memory unit size, default 1kb canal.instance.memory.buffer.memunit = 1024 # # meory store gets mode used MEMSIZE or ITEMSIZE canal.instance.memory.batch.mode = MEMSIZE # # detecing config canal.instance.detecting.enable = false # canal.instance.detecting.sql = insert into retl.xdual values (1 Now () on duplicate key update x=now () canal.instance.detecting.sql = select 1 canal.instance.detecting.interval.time = 3 canal.instance.detecting.retry.threshold = 3 canal.instance.detecting.heartbeatHaEnable = false # support maximum transaction size More than the size of the transaction will be cut into multiple transactions delivery canal.instance.transaction.size = 1024 # mysql fallback connected to new master should fallback times canal.instance.fallbackIntervalInSeconds = 60 # network config canal.instance.network.receiveBufferSize = 16384 canal.instance.network.sendBufferSize = 16384 canal.instance.network.soTimeout = 30 # binlog filter config canal.instance.filter.query.dcl = false canal.instance.filter.query.dml = false canal.instance.filter.query.ddl = false canal.instance.filter.table.error = false Canal.instance.filter.rows = false # binlog format/image check canal.instance.binlog.format = ROW STATEMENT,MIXED canal.instance.binlog.image = FULL,MINIMAL NOBLOB # binlog ddl isolation canal.instance.get.ddl.isolation = false # destinations # # # canal.destinations= example # conf root dir canal.conf.dir =.. / conf # auto scan instance dir add/remove and start/stop instance canal.auto.scan = true canal.auto.scan.interval = 5 canal.instance.global.mode = spring canal.instance.global.lazy = false # canal.instance.global.manager.address = 127.0.0.1 conf 1099 # Canal.instance.global.spring.xml = classpath:spring/memory-instance.xml canal.instance.global.spring.xml = classpath:spring/file-instance.xml # canal.instance.global.spring.xml = classpath:spring/default-instance.xml # mysql serverId canal.instance.mysql.slaveId = 1234 # position info Need to change to your own database information canal.instance.master.address = 127.0.0.1 canal.instance.master.position 3306 canal.instance.master.journal.name = canal.instance.master.timestamp = # canal.instance.standby.address = # canal.instance.standby.journal.name = # canal.instance.standby.timestamp = # username/password You need to change your database information canal.instance.dbUsername = root canal.instance.dbPassword = 123456 canal.instance.defaultDatabaseName = datamip canal.instance.connectionCharset = UTF-8 # table regex canal.instance.filter.regex =. *\.. * #

As it is an overall configuration, you should pay attention to the above three red marks:

The server port number of the current canal of canal.port= 11111

Canal.destinations= example currently has an instance instance named example enabled by default. If you want to open multiple instance, just separate them with "," commas.

Canal.instance.filter.regex =. All tables of all db under the mysql instance are within the scope of monitoring.

2) instance.properties

This is the configuration of a specific instances instance. All configurations that are not involved will be inherited from the canal.properties.

# mysql serverId canal.instance.mysql.slaveId = 1234 # position info canal.instance.master.address = 192.168.23.1 canal.instance.master.journal.name 3306 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 = datamip canal.instance.connectionCharset = UTF-8 # table regex canal.instance.filter.regex =. *\.. * # table black regex canal.instance.filter.black.regex = # # #

Just pay attention to the place marked red above. When you steal binlog, you need to know the mysql address and user name and password.

7. Turn on canal

Remember to configure the / canal/bin directory to the Path of / etc/profile for easy and quick opening. Through the following figure, you will see that port 11111 is already open on centos.

[root@localhost bin] # ls canal.pid startup.bat startup.sh stop.sh [root@localhost bin] # pwd / usr/myapp/canal/bin [root@localhost example] # startup.sh cd to / usr/myapp/canal/bin for workaround relative path LOG CONFIGURATION: / usr/myapp/canal/bin/../conf/logback.xml canal conf:/usr / myapp/canal/bin/../conf/canal.properties 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: cd to / usr/myapp/canal/conf/example for continue [root@localhost example] # netstat-tln Active Internet connections (only servers) Proto Recv-Q Send-Q Local Address Foreign Address State tcp 0 0 0.0 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 0. 01 of LISTEN tcp. 0.0.0.0 LISTEN tcp * LISTEN tcp 0 192.168.122.1 LISTEN tcp 53 0.0.0.0 * LISTEN tcp 00 0.0.0.0 22 0.0.0.0 : 631 0.0.0.0 LISTEN tcp * LISTEN tcp6 0 0 127.0.0.1 LISTEN tcp6 25 0.0.0.0 LISTEN tcp6 0 0: 111: 22 : * LISTEN tcp6 0 0:: 1 LISTEN 631: * LISTEN tcp6 0 0:: 1:25:: * LISTEN [root@localhost example] #

8. Java Client code

Canal driver needs to be fetched in the maven repository: http://www.mvnrepository.com/artifact/com.alibaba.otter/canal.client/1.0.24, but there are still a lot of dependencies.

Com.alibaba.otter canal.client 1.0.24

9. Start the java code for verification

The following code makes a basic judgment on the CURD of table to see if it can be intelligently perceived, and then update the redis according to the actual situation.

Package com.datamip.canal; import java.awt.Event; import java.net.InetSocketAddress; import java.util.List; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; 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.Header Import com.alibaba.otter.canal.protocol.CanalEntry.RowChange; import com.alibaba.otter.canal.protocol.Message; import com.google.protobuf.InvalidProtocolBufferException; public class App {public static void main (String [] args) throws InterruptedException {/ / first step: connect with canal CanalConnector connector = CanalConnectors.newSingleConnector (new InetSocketAddress ("192.168.23.170", 11111), "example", ",") Connector.connect (); / / second step: open subscription connector.subscribe (); / / third step: subscribe to while (true) {try {/ / read 1000 messages at a time Message message = connector.getWithoutAck (1000); long batchID = message.getId () Int size = message.getEntries () .size (); if (batchID = =-1 | | size = = 0) {System.out.println ("No data currently"); Thread.sleep (1000) / / No data} else {System.out.println ("- have data--"); PrintEntry (message.getEntries ()) } / / position id ack (easy to process the next item) connector.ack (batchID);} catch (Exception e) {/ / TODO: handle exception} finally {Thread.sleep (1000) } / / get each printed record @ SuppressWarnings ("static-access") public static void PrintEntry (List entrys) {for (Entry entry: entrys) {/ / first step: disassemble the entry entity Header header = entry.getHeader (); EntryType entryType = entry.getEntryType () / / step 2: if it is RowData, that is the data I need if (entryType = = EntryType.ROWDATA) {String tableName = header.getTableName (); String schemaName = header.getSchemaName (); RowChange rowChange = null; try {rowChange = RowChange.parseFrom (entry.getStoreValue ()) } catch (InvalidProtocolBufferException e) {e.printStackTrace ();} EventType eventType = rowChange.getEventType (); System.out.println (String.format ("currently operating% s, Action=% s", schemaName, tableName, eventType) / / if it is a 'query' or a 'DDL' operation, then sql types out if (eventType = = EventType.QUERY | | rowChange.getIsDdl ()) {System.out.println ("rowchange sql->" + rowChange.getSql ()); return } / / step 3: trace to columns level rowChange.getRowDatasList () .forEach ((rowData)-> {/ / get column situation before update List beforeColumns = rowData.getBeforeColumnsList () / / get the updated column List afterColumns = rowData.getAfterColumnsList (); / / the deletion operation if (eventType = = EventType.DELETE) {PrintColumn (beforeColumns) is currently performed } / / currently performing the insert operation if (eventType = = eventType.INSERT) {PrintColumn (afterColumns) } / / currently performing the update operation if (eventType = = eventType.UPDATE) {PrintColumn (afterColumns);}}) } / / changes to each column on each row public static void PrintColumn (List columns) {columns.forEach ((column)-> {String columnName = column.getName (); String columnValue = column.getValue (); String columnType = column.getMysqlType (); boolean isUpdated = column.getUpdated ()) / / determine whether the field is updated with System.out.println (String.format ("columnName=%s, columnValue=%s, columnType=%s, isUpdated=%s", columnName, columnValue, columnType, isUpdated);});}}

Update operation

Insert operation

Delete operation

From the results, there is nothing wrong, there is a picture and the truth, all right, that's all for this article, it must be helpful for you who develop it.

After reading the above, do you have any further understanding of cache consistency and data heterogeneity solutions for cross-server queries? If you want to know more knowledge or related content, please follow the industry information channel, thank you for your support.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.