spring cloud集成canal

前提

win运行canal

加入canal依赖

<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.3</version>
</dependency>

把ip、端口、监听表名做成配置文件

代码实现

package com.frame.modules.dabis.archives.thread;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.frame.solr.em.SolrCode;
import com.frame.utils.PropertiesLoader;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author liwei
 * @date 2019/8/2 14:39
 * @desc Created with IntelliJ IDEA.
 */
public class CanalThread implements Runnable {

    Log log = LogFactory.getLog(CanalThread.class);

    private String solrName = SolrCode.ARCHIVES.getValue();


    @Override
    public void run() {
        PropertiesLoader loader = new PropertiesLoader("solrConfig.properties");
        listener(loader.getProperty("canalHost"), loader.getProperty("canalPort"), loader.getProperty("canalTable"));
    }


    public void listener(String canalHost, String canalPort, String table) {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalHost, Integer.valueOf(canalPort)), "example", "", "");
        int batchSize = 1000;
        try {
            // 连接
            connector.connect();
            // 监听表
            connector.subscribe(table);
            connector.rollback();
            // 一直循环监听
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                if(-1 != batchId && 0 != message.getEntries().size()) {
                    printEntry(message.getEntries());
                }
                // 提交确认
                connector.ack(batchId);
            }
        } finally {
            connector.disconnect();
        }
    }

    /**
     * 打印具体变化
     * @param entrys
     */
    private void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (CanalEntry.EntryType.TRANSACTIONBEGIN.equals(entry.getEntryType()) || CanalEntry.EntryType.TRANSACTIONEND.equals(entry.getEntryType())) {
                continue;
            }

            CanalEntry.RowChange rowChage = null;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            CanalEntry.EventType eventType = rowChage.getEventType();
            System.out.println(String.format("================> binlog[%s:%s] , 数据库：%s，表名%s , 类型: %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());
                } else {
                    System.out.println("-------修改之前");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------修改之后");
                    printColumn(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private void printColumn(List<CanalEntry.Column> columns) {
        Map<String,Object> aaMap = new HashMap<>();
        for (CanalEntry.Column column : columns) {
            aaMap.put(column.getName(), column.getValue());
        }
        System.out.println( new JSONObject(aaMap).toJSONString());
    }
}

启动线程

新增

 

修改

 

 

删除

 

 

注意：拿到的值都是字符串，建议拿到id反查数据库，拿到对象再同步到自己的缓存。