推荐订阅源

1、下载canal

 下载地址：https://github.com/alibaba/canal/releases 下载其中的开发版本 如： canal.deployer-1.1.8.tar.gz

2、解析修改配置

  2.1 canal.properties

      调整自己的数据推送地方

# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
canal.serverMode = tcp

   2.2 instance.properties  

    调整数据库binlog日志的数据库

3、启动

  startup.bat

注意：需要把plugin中的需要的java 拷贝如lib目录中

4、如果canal.serverMode = tcp，那么就可以自己写一个java对接服务

4.1引入依赖

 <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

        <!-- Canal 核心客户端 -->
        <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>canal.client</artifactId>
            <version>1.1.8</version>
        </dependency>

        <!--协议包 -->
        <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>canal.protocol</artifactId>
            <version>1.1.8</version>
        </dependency>
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
            <version>3.21.12</version>
        </dependency>

4.2、配置yml

canal:
  host: 127.0.0.1
  port: 11111
  destination: example
  username:
  password:

4.3

package com.sleep.canalspringboot.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.net.InetSocketAddress;

@Configuration
public class CanalClientConfig {

    @Value("${canal.host}")
    private String host;
    @Value("${canal.port}")
    private int port;
    @Value("${canal.destination}")
    private String destination;
    @Value("${canal.username}")
    private String username;
    @Value("${canal.password}")
    private String password;

    @Bean
    public CanalConnector canalConnector() {
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), destination, username, password);
        canalConnector.connect(); // 建立连接
        canalConnector.subscribe(); // 订阅数据变更

        return canalConnector;
    }
}

4.4 服务

package com.sleep.canalspringboot.service;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Service
public class CanalDataService {

    @Autowired
    private CanalConnector canalConnector;

    private boolean running = true;
    private Thread thread;

    private static final int BATCH_SIZE = 1000;
    private static final long timeout = 3;

    @PostConstruct
    public void init() {
        thread = new Thread(() -> {
            while (running) {
                try {
                    Message message = canalConnector.get(BATCH_SIZE, timeout, TimeUnit.SECONDS);
                    process(message);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        thread.start();
    }

    private void process(Message message) throws Exception {
        List<Entry> entries = message.getEntries();
        for (Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                String schemaName = entry.getHeader().getSchemaName();
                String tableName = entry.getHeader().getTableName();
                CanalEntry.EventType eventType = rowChange.getEventType();
                List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                // 处理 INSERT/UPDATE/DELETE 等操作
                switch (eventType) {
                    case INSERT:
                        System.out.println("INSERT: " + rowDatasList);
                        for (CanalEntry.RowData rowData : rowDatasList) {
                            System.out.printf("row" + "Data:" + rowData);
                        }
                        break;
                    case UPDATE:
                        System.out.println("UPDATE: " + rowDatasList);
                        for (CanalEntry.RowData rowData : rowDatasList) {
                            System.out.printf("row" + "Data:" + rowData);
                        }
                        break;
                    case DELETE:
                        System.out.println("DELETE: " + rowDatasList);
                        for (CanalEntry.RowData rowData : rowDatasList) {
                            System.out.printf("row" + "Data:" + rowData);
                        }
                        break;
                    default:
                        break;
                }
            }
        }
    }
}