Implementation solution for MySQL data change monitoring

1. Introduction to binlog

1.1 What is binlog?

binlog（Binary Log）Yes MySQL recordDDL(Data definition language, such asCREATE、ALTER)andDML(Data operation language, such asINSERT、UPDATE、DELETE) log file, it is used for:

Master-slave copy: The MySQL master library transmits binlog to the slave library to realize data synchronization.
Data recovery:passmysqlbinlogTool parsing binlog to recover data.
Data synchronization: Third-party tools (such as Canal) parse binlog for data synchronization.

1.2 Three formats of binlog

binlog format	illustrate
STATEMENT	Record the SQL statement itself
ROW	Record line data changes (recommended)
MIXED	Combining the first two, MySQL automatically judges

becauseROWThe format provides accurate row-level change information, so it is recommended.

2. Turn on binlog and configure MySQL

2.1 Check whether binlog is enabled

SHOW VARIABLES LIKE 'log_bin';

iflog_binThe value isOFF, indicating that binlog is not enabled.

2.2 Modify MySQL configuration file (or)

exist[mysqld]The following content was added in part:

server-id=1
log-bin=mysql-bin
binlog-format=ROW
binlog-row-image=FULL
expire_logs_days=7

Restart MySQL:

systemctl restart mysql  # Linux
net stop mysql && net start mysql  # Windows

2.3 Verify binlog configuration

implement:

SHOW BINARY LOGS;

If there is a binlog file, such asmysql-bin.000001, indicating that it is enabled.

3. Listen to binlog using Java

3.1 Selection tool: Canal

Alibaba open sourceCanalIt can simulate the MySQL slave protocol, parse binlog and push incremental data in real time.

3.2 Java code listening binlog

Introducing Maven dependencies

<dependencies>
    <dependency>
        <groupId></groupId>
        <artifactId></artifactId>
        <version>1.1.6</version>
    </dependency>
</dependencies>

Writing Java code

import ;
import ;
import ;
import ;

import ;
import ;

public class BinlogListener {
    public static void main(String[] args) {
        // Connect Canal        CanalConnector connector = (
                new InetSocketAddress("127.0.0.1", 11111), 
                "example", "canal", "canal");
        

        try {
            ();
            (".*\\..*"); // Listen to all library tables            ();
    
            while (true) {
                Message message = (100); // Get data                long batchId = ();
                List&lt;&gt; entries = ();
    
                if (batchId != -1 &amp;&amp; !()) {
                    for ( entry : entries) {
                        if (() == ) {
                            processEntry(entry);
                        }
                    }
                }
                (batchId); // Confirm message            }
        } finally {
            ();
        }
    }
    
    private static void processEntry( entry) {
        try {
             rowChange = (());
             eventType = ();
    
            ("Change Table:" + ().getTableName());
            ("Change Type:" + eventType);
    
            for ( rowData : ()) {
                if (eventType == ) {
                    ("Delete data:" + ());
                } else if (eventType == ) {
                    ("New data:" + ());
                } else {
                    ("Pre-update data:" + ());
                    ("Updated data:" + ());
                }
            }
        } catch (Exception e) {
            ();
        }
    }

}

4. Code parsing

Create a Canal connection

CanalConnector connector = (
    new InetSocketAddress("127.0.0.1", 11111), 
    "example", "canal", "canal");

- 127.0.0.1: Canal Server Address
- 11111: Canal port
- example: Canal instance
- canal/canal: Default account password
Get binlog change data

Message message = (100);

- getWithoutAck(100): Pull 100 binlog events.
Analyze binlog

for ( entry : entries) {
    if (() == ) {
        processEntry(entry);
    }
}

Process onlyROWDATAType changes, ignoring transactions and other information.
Classification processingINSERT、UPDATE、DELETE

if (eventType == ) {
    ("Delete data:" + ());
} else if (eventType == ) {
    ("New data:" + ());
} else {
    ("Pre-update data:" + ());
    ("Updated data:" + ());
}

Summarize

MySQL binlogRecord database changes, which can be used to monitor incremental data.
CanalAs MySQL slave library parses binlog to achieve data synchronization.
Java code examplesShow how to listen with CanalINSERT、UPDATE、DELETEOperation and parse the changed data.

This solution is suitable forDistributed data synchronization、Cache consistencyandData change notification, is an important means of real-time data processing.

The above is the detailed content of the implementation plan for MySQL data change monitoring. For more information about MySQL data change monitoring, please pay attention to my other related articles!