在Java程序中监听mysql的binlog

转载作者：撒哈拉更新时间：2024-10-12 10:26:49

1、背景
2、mysql-binlog-connector-java简介
3、准备工作
- 1、验证数据库是否开启binlog
- 2、开启数据库的binlog
- 3、创建具有REPLICATION SLAVE权限的用户
- 4、事件类型 eventType 解释
  - 1、TABLE_MAP 的注意事项
  - 2、获取操作的列名
- 5、监听binlog的position
  - 1、从最新的binlog位置开始监听
  - 2、从指定的位置开始监听
  - 3、断点续传
  - 6、创建表和准备测试数据
4、功能实现
- 1、从最新的binlog位置开始监听
  - 1、引入jar包
  - 2、监听binlog数据
  - 3、测试
- 2、获取数据更新具体的列名
  - 2.1 新增common-dbutils依赖用于操作数据库
  - 2.2 监听TABLE_MAP事件，获取数据库和表名
  - 2.3 编写工具类获取表的列名和位置信息
  - 2.4 以更新语句为例获取更新的列名和对应的值
    - 1、编写java代码获取更新后的列和值信息
    - 2、执行更新语句
    - 3、查看监听到更新数据信息
- 3、自定义序列化字段
  - 3.1 自定义更新数据text类型字段的反序列
  - 3.2 注册更新数据的反序列
  - 3.3 更新text类型的字段，看输出的结果
- 4、只订阅感兴趣的事件
- 5、断点续传
  - 5.1 binlog信息持久化
  - 5.2、构建BinaryLogClient时，传递已存在的binlog信息
  - 5.3 更新binlog信息
  - 5.4 演示
5、参考地址

1、背景

最近在开发的过程中遇到这么一个问题，当产生某种类型的工单后，需要实时通知到另外的系统，由另外的系统进行数据的研判操作。由于某种原因，像向消息队列中推送工单消息、或直接调用另外系统的接口、或者部署Cannal 等都不可行，因此此处使用 mysql-binlog-connector-java 这个库来完成数据库binlog的监听，从而通知到另外的系统.

2、mysql-binlog-connector-java简介

mysql-binlog-connector-java是一个Java库，通过它可以实现mysql binlog日志的监听和解析操作。它提供了一系列可靠的方法，使开发者通过监听数据库的binlog日志，来实时获取数据库的变更信息，比如：数据的插入、更新、删除等操作.

github地址 https://github.com/osheroff/mysql-binlog-connector-java 。

3、准备工作

1、验证数据库是否开启binlog

mysql> show variables like '%log_bin%';
+---------------------------------+------------------------------------+
| Variable_name                   | Value                              |
+---------------------------------+------------------------------------+
| log_bin                         | ON                                 |
| log_bin_basename                | /usr/local/mysql/data/binlog       |
| log_bin_index                   | /usr/local/mysql/data/binlog.index |
| log_bin_trust_function_creators | OFF                                |
| log_bin_use_v1_row_events       | OFF                                |
| sql_log_bin                     | ON                                 |
+---------------------------------+------------------------------------+

log_bin 的值为 ON 时，表示开启了binlog 。

2、开启数据库的binlog

# 修改 my.cnf 配置文件
[mysqld]
#binlog日志的基本文件名，需要注意的是启动mysql的用户需要对这个目录(/usr/local/var/mysql/binlog)有写入的权限
log_bin=/usr/local/var/mysql/binlog/mysql-bin
# 配置binlog日志的格式
binlog_format = ROW
# 配置 MySQL replaction 需要定义，不能和已有的slaveId 重复
server-id=1

3、创建具有REPLICATION SLAVE权限的用户

CREATE USER binlog_user IDENTIFIED BY 'binlog#Replication2024!';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'binlog_user'@'%';
FLUSH PRIVILEGES;

使用BinaryLogClient需要的权限

4、事件类型 eventType 解释

注意：不同的mysql版本事件类型可能不同，我们本地是mysql8 。

TABLE_MAP: 在表的 insert、update、delete 前的事件，用于记录操作的数据库名和表名。
EXT_WRITE_ROWS: 插入数据事件类型，即 insert 类型
EXT_UPDATE_ROWS: 插入数据事件类型，即 update 类型
EXT_DELETE_ROWS: 插入数据事件类型，即 delete 类型

ROTATE: 当mysqld切换到新的二进制日志文件时写入。当发出一个FLUSH LOGS 语句。或者当前二进制日志文件超过max_binlog_size。

1、TABLE_MAP 的注意事项

一般情况下，当我们向数据库中执行insert、update或delete事件时，一般会先有一个TABLE_MAP事件发出，通过这个事件，我们就知道当前操作的是那个数据库和表。但是如果我们操作的表上存在触发器时，那么可能顺序就会错乱，导致我们获取到错误的数据库名和表名.

2、获取操作的列名

此处以 EXT_UPDATE_ROWS 事件为列，当我们往数据库中update一条记录时，触发此事件，事件内容为

Event{header=EventHeaderV4{timestamp=1727498351000, eventType=EXT_UPDATE_ROWS, serverId=1, headerLength=19, dataLength=201, nextPosition=785678, flags=0}, data=UpdateRowsEventData{tableId=264, includedColumnsBeforeUpdate={0, 1, 2, 3, 4, 5, 6, 7}, includedColumns={0, 1, 2, 3, 4, 5, 6, 7}, rows=[
    {before=[1, zhangsan, 张三-update, 0, [B@7b720427, [B@238552f, 1727524798000, 1727495998000], after=[1, zhangsan, 张三-update, 0, [B@21dae489, [B@2c0fff72, 1727527151000, 1727498351000]}
]}}

从上面的语句中可以看到includedColumnsBeforeUpdate和includedColumns这2个字段表示更新前的列名和更新后的列名，但是这个时候展示的数字，那么如果展示具体的列名呢？可以通过information_schema.COLUMNS获取.

列名获取

5、监听binlog的position

1、从最新的binlog位置开始监听

默认情况下，就是从最新的binlog位置开始监听.

BinaryLogClient client = new BinaryLogClient(hostname, port, username, password);

2、从指定的位置开始监听

BinaryLogClient client = new BinaryLogClient(hostname, port, username, password);
// binlog的文件名
client.setBinlogFilename("");
// binlog的具体位置
client.setBinlogPosition(11);

3、断点续传

这个指的是，当我们的 mysql-binlog-connector-java 程序宕机后，如果数据发生了binlog的变更，我们应该从程序上次宕机的位置的position进行监听，而不是程序重启后从最新的binlog position位置开始监听。默认情况下mysql-binlog-connector-java程序没有为我们实现，需要我们自己去实现。大概的实现思路为:

监听 ROTATE事件，可以获取到最新的binlog文件名和位置。
记录每个事件的position的位置。

6、创建表和准备测试数据

CREATE TABLE `binlog_demo`
(
    `id`          int NOT NULL AUTO_INCREMENT COMMENT '主键',
    `user_name`   varchar(64) DEFAULT NULL COMMENT '用户名',
    `nick_name`   varchar(64) DEFAULT NULL COMMENT '昵称',
    `sex`         tinyint     DEFAULT NULL COMMENT '性别 0-女 1-男 2-未知',
    `address`     text COMMENT '地址',
    `ext_info`    json        DEFAULT NULL COMMENT '扩展信息',
    `create_time` datetime    DEFAULT NULL COMMENT '创建时间',
    `update_time` timestamp NULL DEFAULT NULL COMMENT '修改时间',
    PRIMARY KEY (`id`),
    UNIQUE KEY `uidx_username` (`user_name`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='测试binlog'

-- 0、删除数据
truncate table binlog_demo;

-- 1、添加数据
insert into binlog_demo(user_name, nick_name, sex, address, ext_info, create_time, update_time)
values ('zhangsan', '张三', 1, '地址', '[
  "aaa",
  "bbb"
]', now(), now());

-- 2、修改数据
update binlog_demo
set nick_name   = '张三-update',
    sex         = 0,
    address     = '地址-update',
    ext_info    = '{
      "ext_info": "扩展信息"
    }',
    create_time = now(),
    update_time = now()
where user_name = 'zhangsan';

-- 3、删除数据
delete
from binlog_demo
where user_name = 'zhangsan';

4、功能实现

通过mysql-binlog-connector-java库，当数据库中的表数据发生变更时，进行监听.

1、从最新的binlog位置开始监听

1、引入jar包

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <!-- 监听 mysql binlog -->
    <dependency>
        <groupId>com.zendesk</groupId>
        <artifactId>mysql-binlog-connector-java</artifactId>
        <version>0.29.2</version>
    </dependency>
</dependencies>

2、监听binlog数据

package com.huan.binlog;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.Event;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 初始化 binary log client
 *
 * @author huan.fu
 * @date 2024/9/22 - 16:23
 */
@Component
public class BinaryLogClientInit {

    private static final Logger log = LoggerFactory.getLogger(BinaryLogClientInit.class);

    private BinaryLogClient client;

    @PostConstruct
    public void init() throws IOException, TimeoutException {
        /**
         * # 创建用户
         * CREATE USER binlog_user IDENTIFIED BY 'binlog#Replication2024!';
         * GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'binlog_user'@'%';
         * FLUSH PRIVILEGES;
         */
        String hostname = "127.0.0.1";
        int port = 3306;
        String username = "binlog_user";
        String password = "binlog#Replication2024!";
        // 创建 BinaryLogClient客户端
        client = new BinaryLogClient(hostname, port, username, password);
        // 这个 serviceId 不可重复
        client.setServerId(12);

        // 反序列化配置
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(
                // 将日期类型的数据反序列化成Long类型
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG
        );

        client.setEventDeserializer(eventDeserializer);
        client.registerEventListener(new BinaryLogClient.EventListener() {
            @Override
            public void onEvent(Event event) {
                EventType eventType = event.getHeader().getEventType();
                log.info("接收到事件类型: {}", eventType);
                log.warn("接收到的完整事件: {}", event);
                log.info("============================");
            }
        });
        client.registerLifecycleListener(new BinaryLogClient.AbstractLifecycleListener() {
            @Override
            public void onConnect(BinaryLogClient client) {
                log.info("客户端连接到 mysql 服务器 client: {}", client);
            }

            @Override
            public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
                log.info("客户端和 mysql 服务器 通讯失败 client: {}", client);
            }

            @Override
            public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
                log.info("客户端序列化失败 client: {}", client);
            }

            @Override
            public void onDisconnect(BinaryLogClient client) {
                log.info("客户端断开 mysql 服务器链接 client: {}", client);
            }
        });
        // client.connect 在当前线程中进行解析binlog，会阻塞当前线程
        // client.connect(xxx) 会新开启一个线程，然后在这个线程中解析binlog
        client.connect(10000);
    }

    @PreDestroy
    public void destroy() throws IOException {
        client.disconnect();
    }
}

3、测试

从上图中可以看到，我们获取到了更新后的数据，但是具体更新了哪些列名这个我们是不清楚的.

2、获取数据更新具体的列名

此处以更新数据为例，大体的实现思路如下:

通过监听 TABLE_MAP 事件，用于获取到 insert、update或delete语句操作前的数据库和表。
通过查询 information_schema.COLUMNS 表获取某个表在某个数据库中具体的列信息（比如：列名、列的数据类型等操作）。

2.1 新增common-dbutils依赖用于操作数据库

<!-- 操作数据库 -->
<dependency>
    <groupId>commons-dbutils</groupId>
    <artifactId>commons-dbutils</artifactId>
    <version>1.8.1</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.33</version>
</dependency>

2.2 监听TABLE_MAP事件，获取数据库和表名

定义2个成员变量，database和tableName用于接收数据库和表名。

/**
 * 数据库
 */
private String database;
/**
 * 表名
 */
private String tableName;

监听TABLE_MAP事件，获取数据库和表名

// 成员变量 - 数据库名
private String database;
// 成员变量 - 表名
private String tableName;

client.registerEventListener(new BinaryLogClient.EventListener() {
    @Override
    public void onEvent(Event event) {
        EventType eventType = event.getHeader().getEventType();
        log.info("接收到事件类型: {}", eventType);
        log.info("============================");

        if (event.getData() instanceof TableMapEventData) {
            TableMapEventData eventData = (TableMapEventData) event.getData();
            database = eventData.getDatabase();
            tableName = eventData.getTable();
            log.info("获取到的数据库名: {} 和 表名为: {}", database, tableName);
        }
    }
});

监听TABLE_MAP事件，获取数据库和表名

2.3 编写工具类获取表的列名和位置信息

编写工具类获取表的列名和位置信息

/**
 * 数据库工具类
 *
 * @author huan.fu
 * @date 2024/10/9 - 02:39
 */
public class DbUtils {

    public static Map<String, String> retrieveTableColumnInfo(String database, String tableName) throws SQLException {
        Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/temp_work", "binlog_user", "binlog#Replication2024!");

        QueryRunner runner = new QueryRunner();
        Map<String, String> columnInfoMap = runner.query(
                connection,
                "select a.COLUMN_NAME,a.ORDINAL_POSITION from information_schema.COLUMNS a where a.TABLE_SCHEMA = ? and a.TABLE_NAME = ?",
                resultSet -> {
                    Map<String, String> result = new HashMap<>();
                    while (resultSet.next()) {
                        result.put(resultSet.getString("ORDINAL_POSITION"), resultSet.getString("COLUMN_NAME"));
                    }
                    return result;
                },
                database,
                tableName
        );
        connection.close();
        return columnInfoMap;
    }

    public static void main(String[] args) throws SQLException {
        Map<String, String> stringObjectMap = DbUtils.retrieveTableColumnInfo("temp_work", "binlog_demo");
        System.out.println(stringObjectMap);
    }
}

编写工具类获取表的列名和位置信息

2.4 以更新语句为例获取更新的列名和对应的值

1、编写java代码获取更新后的列和值信息

client.registerEventListener(new BinaryLogClient.EventListener() {
    @Override
    public void onEvent(Event event) {
        EventType eventType = event.getHeader().getEventType();
        log.info("接收到事件类型: {}", eventType);
        log.warn("接收到的完整事件: {}", event);
        log.info("============================");

        // 通过 TableMap 事件获取 数据库名和表名
        if (event.getData() instanceof TableMapEventData) {
            TableMapEventData eventData = (TableMapEventData) event.getData();
            database = eventData.getDatabase();
            tableName = eventData.getTable();
            log.info("获取到的数据库名: {} 和 表名为: {}", database, tableName);
        }

        // 监听更新事件
        if (event.getData() instanceof UpdateRowsEventData) {
            try {
                // 获取表的列信息
                Map<String, String> columnInfo = DbUtils.retrieveTableColumnInfo(database, tableName);
                // 获取更新后的数据
                UpdateRowsEventData eventData = ((UpdateRowsEventData) event.getData());
                // 可能更新多行数据
                List<Map.Entry<Serializable[], Serializable[]>> rows = eventData.getRows();

                for (Map.Entry<Serializable[], Serializable[]> row : rows) {
                    // 更新前的数据
                    Serializable[] before = row.getKey();
                    // 更新后的数据
                    Serializable[] after = row.getValue();
                    // 保存更新后的一行数据
                    Map<String, Serializable> afterUpdateRowMap = new HashMap<>();
                    for (int i = 0; i < after.length; i++) {
                        // 因为 columnInfo 中的列名的位置是从1开始，而此处是从0开始
                        afterUpdateRowMap.put(columnInfo.get((i + 1) + ""), after[i]);
                    }
                    log.info("监听到更新的数据为: {}", afterUpdateRowMap);
                }
            } catch (Exception e) {
                log.error("监听更新事件发生了异常");
            }
        }

        // 监听插入事件
        if (event.getData() instanceof WriteRowsEventData) {
            log.info("监听到插入事件");
        }

        // 监听删除事件
        if (event.getData() instanceof DeleteRowsEventData) {
            log.info("监听到删除事件");
        }
    }
});

2、执行更新语句

update binlog_demo
    set nick_name = '张三-update11',
        -- sex = 0,
        -- address = '地址-update1',
        -- ext_info = '{"ext_info":"扩展信息"}',
        -- create_time = now(),
        update_time = now()
where user_name = 'zhangsan';

3、查看监听到更新数据信息

查看监听到更新数据信息

3、自定义序列化字段

从下图中可知，针对 text 类型的字段，默认转换成了byte[]类型，那么怎样将其转换成String类型呢?

此处针对更新语句来演示。

数据库中text类型变成了字节数组类型

3.1 自定义更新数据text类型字段的反序列

注意：断点跟踪源码发现text类型的数据映射成了blob类型，因此需要重写 deserializeBlob 方法。

public class CustomUpdateRowsEventDataDeserializer extends UpdateRowsEventDataDeserializer {
    public CustomUpdateRowsEventDataDeserializer(Map<Long, TableMapEventData> tableMapEventByTableId) {
        super(tableMapEventByTableId);
    }

    @Override
    protected Serializable deserializeBlob(int meta, ByteArrayInputStream inputStream) throws IOException {
        byte[] bytes = (byte[]) super.deserializeBlob(meta, inputStream);
        if (null != bytes && bytes.length > 0) {
            return new String(bytes, StandardCharsets.UTF_8);
        }
        return null;
    }
}

3.2 注册更新数据的反序列

注意：需要通过 EventDeserializer 来进行注册。

// 反序列化配置
EventDeserializer eventDeserializer = new EventDeserializer();

Field field = EventDeserializer.class.getDeclaredField("tableMapEventByTableId");
field.setAccessible(true);
Map<Long, TableMapEventData> tableMapEventByTableId = (Map<Long, TableMapEventData>) field.get(eventDeserializer);
eventDeserializer.setEventDataDeserializer(EventType.EXT_UPDATE_ROWS, new CustomUpdateRowsEventDataDeserializer(tableMapEventByTableId)
        .setMayContainExtraInformation(true));

3.3 更新text类型的字段，看输出的结果

可以看到数据库中的text类型已经转换成String类型了

4、只订阅感兴趣的事件

// 反序列化配置
EventDeserializer eventDeserializer = new EventDeserializer();
 eventDeserializer.setCompatibilityMode(
         // 将日期类型的数据反序列化成Long类型
         EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG
 );
// 表示对 删除事件不感兴趣 ( 对于DELETE事件的反序列化直接返回null )
 eventDeserializer.setEventDataDeserializer(EventType.EXT_DELETE_ROWS, new NullEventDataDeserializer());

对于不感兴趣的事件直接使用NullEventDataDeserializer，可以提高程序的性能.

5、断点续传

当binlog的信息发生变更时，需要保存起来，下次程序重新启动时，读取之前保存好的binlog信息.

5.1 binlog信息持久化

此处为了模拟，将binlog的信息保存到文件中.

/**
 * binlog position 的持久化处理
 *
 * @author huan.fu
 * @date 2024/10/11 - 12:54
 */
public class FileBinlogPositionHandler {

    /**
     * binlog 信息实体类
     */
    public static class BinlogPositionInfo {
        /**
         * binlog文件的名字
         */
        public String binlogName;
        /**
         * binlog的位置
         */
        private Long position;
        /**
         * binlog的server id的值
         */
        private Long serverId;
    }

    /**
     * 保存binlog信息
     *
     * @param binlogName binlog文件名
     * @param position   binlog位置信息
     * @param serverId   binlog server id
     */
    public void saveBinlogInfo(String binlogName, Long position, Long serverId) {
        List<String> data = new ArrayList<>(3);
        data.add(binlogName);
        data.add(position + "");
        data.add(serverId + "");
        try {
            Files.write(Paths.get("binlog-info.txt"), data);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取 binlog 信息
     *
     * @return BinlogPositionInfo
     */
    public BinlogPositionInfo retrieveBinlogInfo() {
        try {
            List<String> lines = Files.readAllLines(Paths.get("binlog-info.txt"));
            BinlogPositionInfo info = new BinlogPositionInfo();
            info.binlogName = lines.get(0);
            info.position = Long.parseLong(lines.get(1));
            info.serverId = Long.parseLong(lines.get(2));
            return info;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

5.2、构建BinaryLogClient时，传递已存在的binlog信息

// 设置 binlog 信息
FileBinlogPositionHandler fileBinlogPositionHandler = new FileBinlogPositionHandler();
FileBinlogPositionHandler.BinlogPositionInfo binlogPositionInfo = fileBinlogPositionHandler.retrieveBinlogInfo();
if (null != binlogPositionInfo) {
    log.info("获取到了binlog 信息 binlogName: {} position: {} serverId: {}", binlogPositionInfo.binlogName,
            binlogPositionInfo.position, binlogPositionInfo.serverId);
    client.setBinlogFilename(binlogPositionInfo.binlogName);
    client.setBinlogPosition(binlogPositionInfo.position);
    client.setServerId(binlogPositionInfo.serverId);
}

5.3 更新binlog信息

// FORMAT_DESCRIPTION（写入每个二进制日志文件前的描述事件） HEARTBEAT（心跳事件）这2个事件不进行binlog位置的记录
if (eventType != EventType.FORMAT_DESCRIPTION && eventType != EventType.HEARTBEAT) {
    // 当有binlog文件切换时产生
    if (event.getData() instanceof RotateEventData) {
        RotateEventData eventData = event.getData();
        // 保存binlog position 信息
        fileBinlogPositionHandler.saveBinlogInfo(eventData.getBinlogFilename(), eventData.getBinlogPosition(), event.getHeader().getServerId());
    } else {
        // 非 rotate 事件，保存位置信息
        EventHeaderV4 header = event.getHeader();
        FileBinlogPositionHandler.BinlogPositionInfo info = fileBinlogPositionHandler.retrieveBinlogInfo();
        long position = header.getPosition();
        long serverId = header.getServerId();
        fileBinlogPositionHandler.saveBinlogInfo(info.binlogName, position, serverId);
    }
}

5.4 演示

启动程序
修改 address 的值为 地址-update2
停止程序
修改address的值为地址-offline-update
启动程序，看能否收到上一步修改address的值为地址-offline-update的事件

5、参考地址

github地址 - https://github.com/osheroff/mysql-binlog-connector-java
maven仓库地址https://mvnrepository.com/artifact/com.zendesk/mysql-binlog-connector-java/0.29.2
TABLE_MAP事件顺序问题. - https://github.com/shyiko/mysql-binlog-connector-java/issues/67
dbutils的官网 - https://commons.apache.org/proper/commons-dbutils/examples.html

最后此篇关于在Java程序中监听mysql的binlog的文章就讲到这里了,如果你想了解更多关于在Java程序中监听mysql的binlog的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！。

文章推荐：推荐一款支持Vue3的管理系统模版：Vue-Vben-Admin

文章推荐：优秀的面试官！通过一个问题考察了所有网络编程知识点

文章推荐：深入解析SpringAI框架：在Java应用中实现智能化交互的关键

文章推荐： MySQL数据备份

java - Java 方法如何检索属于该特定方法的方法对象？ ( java )
我正在编写一个具有以下签名的 Java 方法。 void Logger(Method method, Object[] args); 如果一个方法(例如 ABC() )调用此方法 Logger，它应该
java - (Java) Java 找不到我的图像文件
我是 Java 新手。我的问题是我的 Java 程序找不到我试图用作的图像文件一个 JButton。 (目前这段代码什么也没做，因为我只是得到了想要的外观第一的)。这是我的主课代码: packag
java - java Java 有手动垃圾回收吗？
好的，今天我在接受采访，我已经编写 Java 代码多年了。采访中说“Java 垃圾收集是一个棘手的问题，我有几个 friend 一直在努力弄清楚。你在这方面做得怎么样？”。她是想骗我吗？还是我的一生都
java - Java 之谜 - Java
我的 friend 给了我一个谜语让我解开。它是这样的: There are 100 people. Each one of them, in his turn, does the following
java - Java 字节码是否兼容不同版本的 Java？
如果我将使用 Java 5 代码的应用程序编译成字节码，生成的 .class 文件是否能够在 Java 1.4 下运行？如果后者可以工作并且我正在尝试在我的 Java 1.4 应用程序中使用 Jav
java - Java 缺少无符号原始类型是 Java 平台的特征还是 Java 语言的特征？
有关于why Java doesn't support unsigned types的问题以及一些关于处理无符号类型的问题。我做了一些搜索，似乎 Scala 也不支持无符号数据类型。限制是Java和S
java - Java 7 的 Java 字节码可以在其他版本的 Java 中工作吗
我只是想知道在一个 java 版本中生成的字节码是否可以在其他 java 版本上运行最佳答案通常，字节码无需修改即可在较新版本的 Java 上运行。它不会在旧版本上运行，除非您使用特殊参数 (
java -cp 。 test.java 与 java test.java
我有一个关于在命令提示符下执行 java 程序的基本问题。在某些机器上我们需要指定 -cp 。 (类路径)同时执行java程序 (test为java文件名与.class文件存在于同一目录下) jav
java - 使用 Java (Java EE/Java SE) 的数据库应用程序设计模式
我已经阅读 StackOverflow 有一段时间了，现在我才鼓起勇气提出问题。我今年 20 岁，目前在我的家乡(罗马尼亚克卢日-纳波卡)就读 IT 大学。足以介绍:D。基本上，我有一家提供簿记应用
java - Java 中的解析可在 Java 中访问
我有 public JSONObject parseXML(String xml) { JSONObject jsonObject = XML.toJSONObject(xml); r
java - Java 中的解释性语言以及对 Java 方法的调用
我已经在 Java 中实现了带有动态类型的简单解释语言。不幸的是我遇到了以下问题。测试时如下代码: def main() { def ks = Map[[1, 2]].keySet()
java - java 序数 - Java I 类
一直提示输入 1 到 10 的数字 - 结果应将 st、rd、th 和 nd 添加到数字中。编写一个程序，提示用户输入 1 到 10 之间的任意整数，然后以序数形式显示该整数并附加后缀。 public
java - 如何从 Java 执行 Java？
我有这个 DownloadFile.java 并按预期下载该文件: import java.io.*; import java.net.URL; public class DownloadFile {
java - 延迟不适用于 java gui(java)
我想在 GUI 上添加延迟。我放置了 2 个 for 循环，然后重新绘制了一个标签，但这 2 个 for 循环一个接一个地执行，并且标签被重新绘制到最后一个。我能做什么？ for(int i=0;
java - Java 类中的硬编码 Java 列表
我正在对对象 Student 的列表项进行一些测试，但是我更喜欢在 java 类对象中创建硬编码列表，然后从那里提取数据，而不是连接到数据库并在结果集中选择记录。然而，自从我这样做以来已经很长时间了，
java - java 幕后对象创建(java 对象实例化)
我知道对象创建分为三个部分: 声明实例化初始化 classA{} classB extends classA{} classA obj = new classB(1,1); 实例化它必须使用
java - 车辆跟踪系统[java/Java EE]
我有兴趣使用 GPRS 构建车辆跟踪系统。但是，我有一些问题要问以前做过此操作的人: GPRS 是最好的技术吗？人们意识到任何问题吗？我计划使用 Java/Java EE - 有更好的技术吗？如果
java - 逆数组(Java)//逆数组(Java)
我可以通过递归方法反转数组，例如:数组={1,2,3,4,5} 数组结果={5,4,3,2,1}但我的结果是相同的数组，我不知道为什么，请帮助我。 public class Recursion { p
java - Java/Java EE 的构建和集成环境
有这样的标准方式吗？包括 Java源代码-测试代码- Ant 或 Maven联合单元持续集成(可能是巡航控制)ClearCase 版本控制工具部署到应用服务器最后我希望有一个自动构建和集成环境。
java - 我将如何从 java 程序打印文本？ ( java )
我什至不知道这是否可能，我非常怀疑它是否可能，但如果可以，您能告诉我怎么做吗？我只是想知道如何从打印机打印一些文本。有什么想法吗？最佳答案这里有更简单的事情。 import javax.swin

撒哈拉

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

在Java程序中监听mysql的binlog

1、背景

2、mysql-binlog-connector-java简介

3、准备工作

1、验证数据库是否开启binlog

2、开启数据库的binlog

3、创建具有REPLICATION SLAVE权限的用户

4、事件类型 eventType 解释

1、TABLE_MAP 的注意事项

2、获取操作的列名

5、监听binlog的position

1、从最新的binlog位置开始监听

2、从指定的位置开始监听

3、断点续传

6、创建表和准备测试数据

4、功能实现

1、从最新的binlog位置开始监听

1、引入jar包

2、监听binlog数据

3、测试

2、获取数据更新具体的列名

2.1 新增common-dbutils依赖用于操作数据库

2.2 监听TABLE_MAP事件，获取数据库和表名

2.3 编写工具类获取表的列名和位置信息

2.4 以更新语句为例获取 更新的列名和对应的值

1、编写java代码获取更新后的列和值信息

2、执行更新语句

3、查看监听到更新数据信息

3、自定义序列化字段

3.1 自定义更新数据text类型字段的反序列

3.2 注册更新数据的反序列

3.3 更新text类型的字段，看输出的结果

4、只订阅感兴趣的事件

5、断点续传

5.1 binlog信息持久化

5.2、构建BinaryLogClient时，传递已存在的binlog信息

5.3 更新binlog信息

5.4 演示

5、参考地址

2.4 以更新语句为例获取更新的列名和对应的值