一、canal安裝

下載地址：https://github.com/alibaba/canal/releases

image.png

二、修改配置文件,監(jiān)聽(tīng)數(shù)據(jù)庫(kù)

默認(rèn)是在example文件夾返吻，可以修改example文件夾中的instance.properties文件配置。當(dāng)然也可以根據(jù)自己的業(yè)務(wù)在conf路徑下新建一個(gè)文件夾乎婿，比如：test,并將example文件夾中的instance.properties文件復(fù)制到test文件夾下测僵。

image.png

image.png

# 找到配置文件此處位置，設(shè)置成自己要監(jiān)聽(tīng)的數(shù)據(jù)庫(kù)地址
canal.instance.master.address=127.0.0.1:3306
# 找到配置文件此處位置谢翎，配置數(shù)據(jù)庫(kù)用戶(hù)名捍靠、密碼、及數(shù)據(jù)庫(kù)名稱(chēng)森逮。
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
canal.instance.defaultDatabaseName = test # test是數(shù)據(jù)庫(kù)名
canal.instance.enableDruid=false
# 找到此處位置榨婆，默認(rèn)是監(jiān)聽(tīng)數(shù)據(jù)庫(kù)下所有的表，可以改成監(jiān)聽(tīng)具體的數(shù)據(jù)表褒侧。
#canal.instance.filter.regex=.*\\..*
canal.instance.filter.regex=test.test_table  # test是數(shù)據(jù)庫(kù)名  test_table是數(shù)據(jù)庫(kù)中表的名稱(chēng)良风，多個(gè)用逗號(hào)隔開(kāi)(test.table1,test.table2)

配置好后，需要修改數(shù)據(jù)庫(kù)的相應(yīng)配置,修改數(shù)據(jù)庫(kù)配置文件my.cnf

log-bin=mysql-bin #添加這一行就ok
binlog-format=ROW #選擇row模式
server_id=1 #配置mysql replaction需要定義闷供，不能和canal的slaveId重復(fù)

配置完成后拖吼，重新啟動(dòng)數(shù)據(jù)庫(kù)，并查看这吻；默認(rèn)是不開(kāi)啟的，為OFF

image.png

配置canal用戶(hù)

CREATE USER canal IDENTIFIED BY 'canal';  # 創(chuàng)建“canal”用戶(hù)
GRANT SELECT,REPLICATION SLAVE,REPLICATION CLIENT ON.TO 'canal'@'%'; # 為“canal”用戶(hù)賦予相應(yīng)權(quán)限
FLUSH PRIVILEGES; # 刷新權(quán)限

配置完成篙议，重啟以下mysql服務(wù)即可唾糯。

三、配置rocketMQ

配置rocketMQ鬼贱，將動(dòng)態(tài)監(jiān)聽(tīng)數(shù)據(jù)庫(kù)發(fā)送的數(shù)據(jù)變化移怯，推送至mq中，并對(duì)相應(yīng)業(yè)務(wù)進(jìn)行處理这难。
修改/conf/canal.properties文件

# 找到以下配置舟误，默認(rèn)的是example文件夾，由于根據(jù)業(yè)務(wù)建立了test文件夾姻乓，所以需要改成test
# canal.destinations = example
canal.destinations = test # 可以配置多個(gè)嵌溢，用逗號(hào)隔開(kāi)眯牧，如：example,test
# 配置rocketMQ
rocketmq.producer.group = test-group  # 根據(jù)自己的業(yè)務(wù)進(jìn)行修改
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic = test_canal_topic # 設(shè)置topic
rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876 # 配置rocketMQ地址
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag = 

配置好rocketMQ后，還需要修改/conf/test/instance.properties文件的topic配置

# 配置mq中的topic
canal.mq.topic=canal_test_topic

四赖草、啟動(dòng)canal

啟動(dòng)/bin/startup.bat即可学少。

五、springboot項(xiàng)目監(jiān)聽(tīng)canal發(fā)送rocketMQ的代碼

springboot中的rocketMQ配置

#rocketmq配置
rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: test-group  #這個(gè)需要和canal中/conf/canal.properties文件中的rocketmq.producer.group配置一致

springboot相應(yīng)代碼

# topic 需要和canal中的/conf/test/instance.properties文件中的canal.mq.topic配置一致
# consumerGroup = "consumer-test-canal-group" # 自定義一個(gè)消費(fèi)組即可
# customizedTraceTopic需要和canal中的/conf/canal.properties文件中的rocketmq.customized.trace.topic配置一致
@Component
@RocketMQMessageListener(topic = "canal_test_topic", consumerGroup = "consumer-test-canal-group", customizedTraceTopic = "test_canal_topic")
public class CanalConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        System.out.println(msg);
        if (!JSONUtil.isJson(msg)) {
            return;
        }
        JSONObject msgJsonObject = JSONUtil.parseObj(msg);
        String sqlType = msgJsonObject.getStr("type");
        switch (sqlType) {
            case "UPDATE":
            case "INSERT":
            case "DELETE":
                break;
            default:
                System.out.println("不同步的消息類(lèi)型：" + sqlType);
                return;
        }
        System.out.println(msgJsonObject.toStringPretty());
        String table = msgJsonObject.getStr("table");  # 獲取表名
        JSONArray data = msgJsonObject.getJSONArray("data");
        JSONArray pkNames = msgJsonObject.getJSONArray("pkNames");
        System.out.println("表名為:" + msgJsonObject.getStr("table") + ",sql類(lèi)型為：" + sqlType);
        if ("UPDATE".equals(sqlType) || "INSERT".equals(sqlType)) {
          #  根據(jù)自己的業(yè)務(wù)進(jìn)行處理
            for (int i = 0; i < data.size(); i++) {
                JSONObject object = data.getJSONObject(i);
                # 可以根據(jù)序號(hào)及字段獲取相應(yīng)信息
                String key = object.getStr(pkNames.getStr(i), "0");
                String id = object.getStr("id","");
            }
            return;
        }
        if ("DELETE".equals(sqlType)) {
           #  根據(jù)自己的業(yè)務(wù)進(jìn)行處理
            for (int i = 0; i < data.size(); i++) {
                JSONObject object = data.getJSONObject(i);
                String key = object.getStr(pkNames.getStr(i), "0");
            }
        }
    }
}