MySQL特定表全量、增量數據同步到消息隊列-解決方案

Posted on by

1、原始需求

既要同步原始全量數據，也要實時同步MySQL特定庫的特定表增量數據，同時對應的修改、刪除也要對應。

數據同步不能有侵入性：不能更改業務程序，並且不能對業務側有太大性能壓力。

應用場景：數據ETL同步、降低業務服務器壓力。

2、解決方案

3、canal介紹、安裝

canal是阿裡巴巴旗下的一款開源項目，純Java開發。基於數據庫增量日志解析，提供增量數據訂閱&消費，目前主要支持瞭MySQL（也支持mariaDB）。

工作原理：mysql主備復制實現

從上層來看，復制分成三步：

  1. master將改變記錄到二進制日志(binary log)中（這些記錄叫做二進制日志事件，binary log events，可以通過show binlog events進行查看）；
  2. slave將master的binary log events拷貝到它的中繼日志(relay log)；
  3. slave重做中繼日志中的事件，將改變反映它自己的數據。

canal的工作原理

原理相對比較簡單：

  1. canal模擬mysql slave的交互協議，偽裝自己為mysql slave，向mysql master發送dump協議
  2. mysql master收到dump請求，開始推送binary log給slave(也就是canal)
  3. canal解析binary log對象(原始為byte流)

架構

說明：

  • server代表一個canal運行實例，對應於一個jvm
  • instance對應於一個數據隊列 （1個server對應1..n個instance)

instance模塊：

  • eventParser (數據源接入，模擬slave協議和master進行交互，協議解析)
  • eventSink (Parser和Store鏈接器，進行數據過濾，加工，分發的工作)
  • eventStore (數據存儲)
  • metaManager (增量訂閱&消費信息管理器)

安裝

1、mysql、kafka環境準備

2、canal下載：wget https://github.com/alibaba/canal/releases/download/canal-1.1.3/canal.deployer-1.1.3.tar.gz

3、解壓：tar -zxvf canal.deployer-1.1.3.tar.gz

4、對目錄conf裡文件參數配置

對canal.properties配置：

進入conf/example裡，對instance.properties配置：

5、啟動：bin/startup.sh

6、日志查看：

4、驗證

1、開發對應的kafka消費者

package org.kafka;

import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;


/**
 *
 * Title: KafkaConsumerTest
 * Description:
 *  kafka消費者 demo
 * Version:1.0.0
 * @author pancm
 * @date 2018年1月26日
 */
public class KafkaConsumerTest implements Runnable {

    private final KafkaConsumer<String, String> consumer;
    private ConsumerRecords<String, String> msgList;
    private final String topic;
    private static final String GROUPID = "groupA";

    public KafkaConsumerTest(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.7.193:9092");
        props.put("group.id", GROUPID);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("auto.offset.reset", "latest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        this.topic = topicName;
        this.consumer.subscribe(Arrays.asList(topic));
    }

    @Override
    public void run() {
        int messageNo = 1;
        System.out.println("---------開始消費---------");
        try {
            for (; ; ) {
                msgList = consumer.poll(1000);
                if (null != msgList && msgList.count() > 0) {
                    for (ConsumerRecord<String, String> record : msgList) {
                        //消費100條就打印 ,但打印的數據不一定是這個規律的

                            System.out.println(messageNo + "=======receive: key = " + record.key() + ", value = " + record.value() + " offset===" + record.offset());


//                            String v = decodeUnicode(record.value());

//                            System.out.println(v);

                        //當消費瞭1000條就退出
                        if (messageNo % 1000 == 0) {
                            break;
                        }
                        messageNo++;
                    }
                } else {
                    Thread.sleep(11);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }

    public static void main(String args[]) {
        KafkaConsumerTest test1 = new KafkaConsumerTest("sample-data");
        Thread thread1 = new Thread(test1);
        thread1.start();
    }


    /*
     * 中文轉unicode編碼
     */
    public static String gbEncoding(final String gbString) {
        char[] utfBytes = gbString.toCharArray();
        String unicodeBytes = "";
        for (int i = 0; i < utfBytes.length; i++) {
            String hexB = Integer.toHexString(utfBytes[i]);
            if (hexB.length() <= 2) {
                hexB = "00" + hexB;
            }
            unicodeBytes = unicodeBytes + "\\u" + hexB;
        }
        return unicodeBytes;
    }

    /*
     * unicode編碼轉中文
     */
    public static String decodeUnicode(final String dataStr) {
        int start = 0;
        int end = 0;
        final StringBuffer buffer = new StringBuffer();
        while (start > -1) {
            end = dataStr.indexOf("\\u", start + 2);
            String charStr = "";
            if (end == -1) {
                charStr = dataStr.substring(start + 2, dataStr.length());
            } else {
                charStr = dataStr.substring(start + 2, end);
            }
            char letter = (char) Integer.parseInt(charStr, 16); // 16進制parse整形字符串。
            buffer.append(new Character(letter).toString());
            start = end;
        }
        return buffer.toString();

    }
}

2、對表bak1進行增加數據

CREATE TABLE `bak1` (
  `vin` varchar(20) NOT NULL,
  `p1` double DEFAULT NULL,
  `p2` double DEFAULT NULL,
  `p3` double DEFAULT NULL,
  `p4` double DEFAULT NULL,
  `p5` double DEFAULT NULL,
  `p6` double DEFAULT NULL,
  `p7` double DEFAULT NULL,
  `p8` double DEFAULT NULL,
  `p9` double DEFAULT NULL,
  `p0` double DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4

show create table bak1;

insert into bak1 select '李雷abcv',
  `p1` ,
  `p2` ,
  `p3` ,
  `p4` ,
  `p5` ,
  `p6` ,
  `p7` ,
  `p8` ,
  `p9` ,
  `p0`  from moci limit 10

3、查看輸出結果：

到此這篇關於MySQL特定表全量、增量數據同步到消息隊列-解決方案的文章就介紹到這瞭,更多相關MySQL特定表數據同步內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：