Binlog4j

轻量级 Mysql Binlog 客户端, 提供宕机续读, 高可用集群等特性

更新日志

[新增] GenericBinlogEventHandler 抽象类。
[新增] BinlogClientConfig 类 setPositionHandler 方法，用于自定义 position 记录逻辑。
[新增] IBinlogClient 类 registerEventHandler(String handlerKey, IBinlogEventHandler handler) 方法。
[新增] IBinlogClient 类 unregisterEventHandler(String handlerKey) 方法。
[优化] Mysql bit 类型数据处理, 兼容 BitSet Boolean boolean Java 类型。
[优化] Mysql text 与 longtext 类型数据处理, 开箱即明文, 不再需要 base64 解码。
[优化] BinlogEventHandler 数据转换 时的性能问题。
[修复] Mysql datetime 类型数据不能为空的问题。
[修复] 首次启动时 binlogPosition 存 0 的问题。
[修复] 缺少 Gson 依赖的异常信息提示。
[破坏] 新增 IBinlogEventHandler 类 isHandle(BinlogEvent event) 接口。
[破坏] 移除 BinlogSubscriber 注解 database 与 table 配置。
[破坏] 移除 IBinlogClient 类 registerEventHandler(String database, String table, .. handler) 方法

项目特性

• 集群模式, 通过集群部署的方式，从而实现服务的高可用。

• 宕机续读, 避免宕机期间造成数据丢失, 保证数据一致性。

• 支持 传统项目 与 Spring Boot 项目集成, 同时兼容 Spring Boot 2.x 与 3.x 版本。

• 数据转换, 基于 IBinlogEventHandler 的泛型参数, 提供自动的数据转换。

项目地址

Gitee 仓库：https://gitee.com/dromara/binlog4j

下载安装

<dependency>
   <groupId>com.gitee.Jmysy</groupId>
   <artifactId>binlog4j-core</artifactId>
   <version>latest.version</version>
</dependency>

或

implementation group: 'com.gitee.Jmysy', name: 'binlog4j-core', version: 'latest.version'

简单使用

通过 BinlogClient 创建 binlog 客户端, IBinlogEventHandler 为 binlog 事件的通知, 配置 host 中的所有数据变化, 都将会被推送到 onUpdate, onDelete, onInsert 接口。 isHandle 为 1.4.0 新增, 用于判定当前 handler 是否需要被执行, 你可以用过 BinlogEvent 获取到 database 和 table 判定依据。

public class BootStrap {

    public static void main(String[] args) {
        
        BinlogClientConfig clientConfig = new BinlogClientConfig();
        clientConfig.setHost("127.0.0.1");
        clientConfig.setPort(3306);
        clientConfig.setUsername("root");
        clientConfig.setPassword("taoren@123");
        clientConfig.setServerId(1990);
  
        IBinlogClient binlogClient = new BinlogClient(clientConfig);

        binlogClient.registerEventHandler(new IBinlogEventHandler() {
            
            @Override
            public void onInsert(BinlogEvent event) {
                System.out.println("插入数据:{}", event.getData());
            }

            @Override
            public void onUpdate(BinlogEvent event) {
                System.out.println("修改数据:{}", event.getData());
            }

            @Override
            public void onDelete(BinlogEvent event) {
                System.out.println("删除数据:{}", event.getData());
            }

            @Override
            public boolean isHandle(BinlogEvent event) {
                return event.getDatabase().equals("pear-admin") && event.getTable().equals("sys_user");
            }
        });

        binlogClient.connect();
    }
}

宕机续读

防止宕机期间的数据丢失, 从而保证数据一致性, persistence[boolean] 为此而生。你可以通过设置 true 来开启, 每次处理 binlog event 时, binlog4j 都将会 当前的消费进度记录到 Redis 中, 从而在下次启动时实现续读。

RedisConfig redisConfig = new RedisConfig();
redisConfig.setHost("127.0.0.1");
redisConfig.setPort(6379);
redisConfig.setPassword("taoren@123");

BinlogClientConfig clientConfig = new BinlogClientConfig();
clientConfig.setRedisConfig(redisConfig); // Redis 配置
clientConfig.setPersistence(true); // 开启续读

PS： persistence 的功能实现依赖于 Redis 中间件, 在开启时, 你需要提供 RedisConfig 的配置。

集群模式

在实际应用场景中, 为了提供服务的可用性, 服务常常以集群的方式呈现。在 binlog4j 中, 你可以通过 mode 配置, 开启 cluster（集群） 模式, 默认为 standalone（单机）。 cluster 模式借助 Redisson 实现, RedisConfig 不可或缺。需要注意的是, 不同的集群你需要保证 serverId 配置的唯一性。

RedisConfig redisConfig = new RedisConfig();
redisConfig.setHost("127.0.0.1");
redisConfig.setPort(6379);
redisConfig.setPassword("taoren@123");

BinlogClientConfig clientConfig = new BinlogClientConfig();
clientConfig.setRedisConfig(redisConfig); // Redis 配置
clientConfig.setMode(BinlogClientMode.cluster); // 集群模式

泛型参数

在 BinlogEvent 中 data 与 originalData 的 Class 类型为 Map<String, Object>, 为进一步降低使用的心智负担, IBinlogEventHandler 接口提供了泛型参数的支持, binlog4j 将依据泛型参数, 将 接收到的数据转换为 JavaBean。

public class UserBinlogEventHandler implements IBinlogEventHandler<User> {

    @Override
    public void onInsert(BinlogEvent<User> event) {
        System.out.println("插入数据：" + event.getData());
    }

    @Override
    public void onUpdate(BinlogEvent<User> event) {
        System.out.println("修改数据:" + event.getData());
    }

    @Override
    public void onDelete(BinlogEvent<User> event) {
        System.out.println("删除数据:" + event.getData());
    }

    @Override
    public boolean isHandle(BinlogEvent<User> event) {
        return event.getDatabase().equals("pear-admin") && event.getTable().equals("sys_user");
    }

}

• bit -> BitSet | Boolean | boolean
• text | longtext -> String
• datetime -> Date | LocalDate | LocalDateTime | LocalTime
• ....

Spring Boot Starter

为了进一步简化 binlog4j 在 Spring Boot 项目中的使用, 我们提供了 binlog4j-spring-boot-starter 包, 它将帮助你在 Spring Boot 中快速集成。

<dependency>
    <groupId>com.gitee.Jmysy</groupId>
    <artifactId>binlog4j-spring-boot-starter</artifactId>
    <version>latest.version</version>
</dependency>

或

implementation group: 'com.gitee.Jmysy', name: 'binlog4j-spring-boot-starter', version: 'latest.version'

首先, 在 application.yml / application.properties 中编写 binlog4j 所需配置

spring:
  binlog4j:
    redis-config:
      host: 127.0.0.1
      port: 6379
      password: taoren@123
    client-configs:
      master:
        username: root
        password: taoren@123
        host: 127.0.0.1
        port: 3306
        serverId: 1990
      slave:
        username: root
        password: taoren@123
        host: 127.0.0.1
        port: 3307
        serverId: 1991

在 Spring Boot 中存在多个 client 的概念，通过 @BinlogSubscriber 注解指定 handler 绑定的客户端, 本质上它代替了 binlogClient.registerEventHandler(IBinlogEventHandler)。

@BinlogSubscriber(clientName = "master")
public class UserEventHandler implements IBinlogEventHandler<User> {

    @Override
    public void onInsert(BinlogEvent<User> event) {
        System.out.println("插入数据：" + event.getData());
    }

    @Override
    public void onUpdate(BinlogEvent<User> event) {
        System.out.println("修改数据:" + event.getData());
    }

    @Override
    public void onDelete(BinlogEvent<User> event) {
        System.out.println("删除数据:" + event.getData());
    }

    @Override
    public boolean isHandle(BinlogEvent<User> event) {
        return event.getDatabase().equals("pear-admin") && event.getTable().equals("sys_user");
    }

}