阿里云DTS同步binlog实战

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
RDS MySQL Serverless 高可用系列,价值2615元额度,1个月
简介: 阿里云DTS同步binlog实战

阿里云DTS同步binlog实战



首先在阿里云DTS订阅中配置你要订阅的库表,其次创建消费者组,最后就是写代码消费binlog即可。


一、创建的配置信息



从这里得到topic以及broker信息。



从这里得到消费者组ID。


二、写代码消费



  1. 创建消费类
import com.aliyun.dts.subscribe.clients.ConsumerContext;
import com.aliyun.dts.subscribe.clients.DTSConsumer;
import com.aliyun.dts.subscribe.clients.DefaultDTSConsumer;
import com.aliyun.dts.subscribe.clients.common.RecordListener;
import com.aliyun.dts.subscribe.clients.record.OperationType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Collections;
import java.util.Map;
@Component
public class DTSConsumerSubscribe {
    // 自己实现binlog解析组件
    @Resource
    MysqlRecordConsumeListener mysqlRecordConsumeListener;
    private static final Logger log = LoggerFactory.getLogger(DTSConsumerSubscribe.class);
    private final DTSConsumer dtsConsumer; // DTS消费实例
    public DTSConsumerSubscribe() {
        // kafka broker url
        String brokerUrl = "xxxxxxx";
        // topic to consume, partition is 0
        String topic = "xxxxxx";
        String sid = "ccc"; // 消费者组ID
        String userName = "xxx"; // 消费者组名称
        String password = "xxx"; // 消费者组密码
        // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
        String initCheckpoint = "1693567696";
        // when use subscribe mode, group config is required. kafka consumer group is enabled
        ConsumerContext.ConsumerSubscribeMode subscribeMode = ConsumerContext.ConsumerSubscribeMode.ASSIGN;
        this.dtsConsumer = initDTSClient(brokerUrl, topic, sid, userName, password, initCheckpoint, subscribeMode);
    }
    public DTSConsumerSubscribe(String brokerUrl, String topic, String sid, String userName, String password,
                                String checkpoint, ConsumerContext.ConsumerSubscribeMode subscribeMode) {
        this.dtsConsumer = initDTSClient(brokerUrl, topic, sid, userName, password, checkpoint, subscribeMode);
    }
    // 初始化DTS客户端
    private DTSConsumer initDTSClient(String brokerUrl, String topic, String sid, String userName, String password,
                                      String initCheckpoint, ConsumerContext.ConsumerSubscribeMode subscribeMode) {
        ConsumerContext consumerContext = new ConsumerContext(brokerUrl, topic, sid, userName, password, initCheckpoint, subscribeMode);
        consumerContext.setForceUseCheckpoint(false);
        DTSConsumer dtsConsumer = new DefaultDTSConsumer(consumerContext);
        dtsConsumer.addRecordListeners(buildRecordListener());
        return dtsConsumer;
    }
    // 构建binlog监听器 
    public Map<String, RecordListener> buildRecordListener() {
        // user can impl their own listener
        RecordListener mysqlRecordPrintListener = record -> {
            OperationType operationType = record.getOperationType();
            if(operationType.equals(OperationType.INSERT)
                    || operationType.equals(OperationType.UPDATE)
                    || operationType.equals(OperationType.DELETE)
                    || operationType.equals(OperationType.DDL)) {
                // consume record
                mysqlRecordConsumeListener.consume(record);
                record.commit(""); // 提交消费位点
            }
        };
        return Collections.singletonMap("mysqlRecordPrinter", mysqlRecordPrintListener);
    }
    // 开启消费
    public void start() {
        System.out.println("Start DTS subscription client...");
        dtsConsumer.start();
    }
    // 关闭消费
    public void stop() {
        System.out.println("Stop DTS subscription client");
        dtsConsumer.close();
    }


  1. 创建自己的解析器
@Component
public class MysqlRecordConsumeListener implements RecordListener {
    private static final Logger log = LoggerFactory.getLogger(MysqlRecordConsumeListener.class);
    public void consume(DefaultUserRecord record) {
        OperationType operationType = record.getOperationType();
        RecordSchema recordSchema = record.getSchema();
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("\n").append("RecordID [").append(record.getId()).append("]\n").append("RecordTimestamp [").append(record.getSourceTimestamp()).append("] \n").append("Source [").append(recordSchema.getDatabaseInfo()).append("]\n").append("RecordType [").append(record.getOperationType()).append("]\n");
        if (operationType.equals(OperationType.INSERT) || operationType.equals(OperationType.UPDATE) || operationType.equals(OperationType.DELETE) || operationType.equals(OperationType.DDL)) {
            stringBuilder.append("Schema info [").append(recordSchema.toString()).append("]\n").append("Before image {").append(record.getBeforeImage()).append("}\n").append("After image {").append(record.getAfterImage()).append("}\n");
            parseRecord(operationType, record);
        }
        log.info(stringBuilder.toString());
    }
    public void parseRecord(OperationType operationType, DefaultUserRecord record) {
        log.info("解析binlog record:{}", record);
        if (OperationType.INSERT.equals(operationType)) { // 新增只有After 对于After,取出自增字段id查询即可
            log.info("=====INSERT getValues=====");
        } else if (OperationType.UPDATE.equals(operationType)) { // 修改有Before和After 取出自增字段id查询即可
            log.info("=====UPDATE getValues=====");
        } else if (OperationType.DELETE.equals(operationType)) { // 删除只有Before 取出自增字段id查询即可
            log.info("=====DELETE getValues=====");
        }
    }}


拿到binlog之后就可以轻松关联相关表做些操作了。大家可以看下UPDATE的binlog的打印日志:


Before image {[Field [id] [4]
Field [user_id] [1002316]
Field [user_name] [ascacac]
Field [goods_id] [1054]
Field [course_id] [null]
Field [course_name] [null]
Field [current_status] [null]
]}
After image {[Field [id] [4]
Field [user_id] [1002316]
Field [user_name] [jamlee]
Field [goods_id] [1054]
Field [course_id] [null]
Field [course_name] [null]
Field [current_status] [null]
]}


  1. 写task跑起来
@Component
public class UserRightDtsSubscribe implements ApplicationRunner, DisposableBean {
    @Resource
    DTSConsumerSubscribe dtsConsumerSubscribe;
    @Override
    public void destroy() throws Exception {
        log.info("关闭用户订单数据同步...");
        dtsConsumerSubscribe.stop();
    }
    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("启动用户订单数据同步...");
        dtsConsumerSubscribe.start();
    }
}


三、总结


阿里云DTS这种订阅消费binlog方式非常方便,尤其在做增量数据同步或者修改等操作的时候。当然也可以通过别的方式同步binlog,比如canal等。但是binlog这种同步方式有缺点吗或者会导致哪些问题呢?

相关实践学习
如何快速连接云数据库RDS MySQL
本场景介绍如何通过阿里云数据管理服务DMS快速连接云数据库RDS MySQL,然后进行数据表的CRUD操作。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助 &nbsp; &nbsp; 相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
打赏
0
0
0
0
5
分享
相关文章
阿里云DTS踩坑经验分享系列|DTS打通SQL Server数据通道能力介绍
SQL Server 以其卓越的易用性和丰富的软件生态系统,在数据库行业中占据了显著的市场份额。作为一款商业数据库,外部厂商在通过解析原生日志实现增量数据捕获上面临很大的挑战,DTS 在 SQL Sever 数据通道上深研多年,提供了多种模式以实现 SQL Server 增量数据捕获。用户可以通过 DTS 数据传输服务,一键打破自建 SQL Server、RDS SQL Server、Azure、AWS等他云 SQL Server 数据孤岛,实现 SQL Server 数据源的流动。
108 0
阿里云DTS踩坑经验分享系列|DTS打通SQL Server数据通道能力介绍
阿里云DTS踩坑经验分享系列|DTS SelectDB链路最佳实践
大数据时代背景下,高效的数据流转与实时分析能力对于企业的竞争力至关重要。阿里云数据传输服务DTS与SelectDB联合,为用户提供了简单、实时、极速且低成本的事务数据分析方案。用户可以通过 DTS 数据传输服务,一键将自建 MySQL/PostgreSQL、RDS MySQL/PostgreSQL、PolarDB for MySQL/PostgreSQL 数据库,迁移或同步至阿里云数据库 SelectDB 的实例中,帮助企业在短时间内完成数据迁移或同步,并即时获得深度洞察。
260 3
阿里云DTS踩坑经验分享系列|SLS同步至ClickHouse集群
作为强大的日志服务引擎,SLS 积累了用户海量的数据。为了实现数据的自由流通,DTS 开发了以 SLS 为源的数据同步插件。目前,该插件已经支持将数据从 SLS 同步到 ClickHouse。通过这条高效的同步链路,客户不仅能够利用 SLS 卓越的数据采集和处理能力,还能够充分发挥 ClickHouse 在数据分析和查询性能方面的优势,帮助企业显著提高数据查询速度,同时有效降低存储成本,从而在数据驱动决策和资源优化配置上取得更大成效。
223 9
阿里云DTS踩坑经验分享系列|使用VPC数据通道解决网络冲突问题
阿里云DTS作为数据世界高速传输通道的建造者,每周为您分享一个避坑技巧,助力数据之旅更加快捷、便利、安全。本文介绍如何使用VPC数据通道解决网络冲突问题。
227 0
Maxwell:binlog 解析器,轻松同步 MySQL 数据
Maxwell:binlog 解析器,轻松同步 MySQL 数据
633 11
阿里云DTS踩坑经验分享系列|Redis迁移、同步
阿里云数据传输服务DTS在帮助用户迁移Redis数据、同步数据时,在某些复杂场景下会出现报错,或者源库与目标库数据不一致的问题,给用户带来困扰。本文介绍了DTS Redis到Redis迁移、同步过程中的典型问题,以帮助用户更好地使用DTS。
501 2
【TiDB原理与实战详解】5、BR 物理备份恢复与Binlog 数据同步~学不会? 不存在的!
BR(Backup & Restore)是 TiDB 分布式备份恢复的命令行工具,适用于大数据量场景,支持常规备份恢复及大规模数据迁移。BR 通过向各 TiKV 节点下发命令执行备份或恢复操作,生成 SST 文件存储数据信息与 `backupmeta` 文件存储元信息。推荐部署配置包括在 PD 节点部署 BR 工具,使用万兆网卡等。本文介绍 BR 的工作原理、部署配置、使用限制及多种备份恢复方式,如全量备份、单库/单表备份、过滤备份及增量备份等。
阿里云DTS踩坑经验分享系列|全量迁移加速方法指南
阿里云数据传输服务DTS是一个便捷、高效的数据迁移和数据同步服务。一般而言,一个完整的DTS数据迁移任务主要包括预检查、结构迁移,全量迁移,增量迁移等阶段,其中全量迁移会将源数据库的存量数据全部迁移到目标数据库。面对各种各样的用户场景, 本文将重点介绍如何使用阿里云DTS实现全量数据迁移加速,以缩短迁移时间,确保数据迁移的效率和稳定性。
782 0
DTS 的惊天挑战:迁移海量 MongoDB 数据时,捍卫数据准确完整的生死之战!
【8月更文挑战第7天】在数字化时代,大数据量的MongoDB迁移至关重要。DTS(数据传输服务)通过全面的数据评估、可靠的传输机制(如事务保证一致性)、异常处理(如回滚或重试),以及迁移后的数据校验来确保数据准确无损。DTS还处理数据转换与映射,即使面对不同数据库结构也能保持数据完整性,为企业提供可靠的数据迁移解决方案。
123 2
MaxCompute产品使用问题之dts是否支持传输数据到mc主键表2.0
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等