How to use Java to implement MySQL data change monitoring

In high concurrency and big data environments, real-time acquisition of incremental changes in MySQL database is crucial for data synchronization, data analysis, cache updates and other scenarios. MySQLbinlog（Binary Log）All changes to the database are recorded and can be used to implementIncremental data monitoring. This article will introduce how to use binlog to listen for MySQL data increments and provide Java-basedCanalImplementation example.

Introduction

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

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 is added in the section:

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<> entries = ();
    
                if (batchId != -1 && !()) {
                    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

1. 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

2. Obtain binlog change data

Message message = (100);

getWithoutAck(100): Pull 100 binlog events.

3. Analyze binlog

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

Process onlyROWDATAChanges in type, ignore transactions and other information.

4. Classification processing INSERT, 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.

This is the end of this article about how to use Java to implement MySQL data change monitoring. For more information about Java monitoring MySQL data changes, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!