canal入门教程

• cannal安装
  下载地址https://github.com/alibaba/canal/releases/tag/canal-1.1.1
  下载后解压安装即可
• 配置
  canal.properties中配置

#mysql主库连接
canal.instance.master.address = 127.0.0.1:3306
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =wb
#可根据自己需求监控要同步的表这里全部监控
#canal.instance.filter.regex = canaldb\\..*
canal.instance.filter.regex = .*\\..*
#监控黑名单配置（不监控）
canal.instance.filter.black.regex = mysql\\..*

canal.instance.parser.parallelThreadSize = 16

instance.properties中配置

#mysql主库连接
canal.instance.master.address = 127.0.0.1:3306

#username/password，需要改成自己的数据库信息
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =canaldb
canal.instance.connectionCharset = UTF-8

• 修改mysql的binlog格式

#查看
SHOW VARIABLES LIKE 'binlog_format';

可以看到，在MySQL中默认的binlog格式为STATEMENT，这里我们需要将STATEMENT修改为ROW。修改/etc/my.cnf文件。
修改mysql配置

vim /etc/my.cnf

在[mysqld]下面新增如下三项配置。

log-bin=mysql-bin  #开启MySQL二进制日志
binlog_format=ROW #将二进制日志的格式设置为ROW
server_id=1 #server_id需要唯一，不能与Canal的slaveId重复

然后重启mysql

service mysqld restart

• 配置mysql权限

#canal和mysql在一台机器的话直接赋予localhost全部权限
CREATE USER canal@'localhost' IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'localhost';
FLUSH PRIVILEGES;

#canal和mysql不在一台机器的话直接赋予%全部权限
CREATE USER canal@'%' IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

• java测试代码

package com.lenovo.test;

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.google.protobuf.InvalidProtocolBufferException;

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

/**
 * Created by chenwei01 on 2017/4/9.
 */
public class CanalClient {

    public static void main(String[] args) {
        while (true) {
            //连接canal
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.19.11", 11111), "example", "canal", "canal");
            connector.connect();
            //订阅 监控的 数据库.表
            connector.subscribe("wb.t_user");
            //一次取5条
            Message msg = connector.getWithoutAck(5);

            long batchId = msg.getId();
            int size = msg.getEntries().size();
            if (batchId < 0 || size == 0) {
                System.out.println("没有消息，休眠5秒");
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                //
                CanalEntry.RowChange row = null;
                for (CanalEntry.Entry entry : msg.getEntries()) {
                    try {
                        row = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                        List<CanalEntry.RowData> rowDatasList = row.getRowDatasList();
                        for (CanalEntry.RowData rowdata : rowDatasList) {
                            List<CanalEntry.Column> afterColumnsList = rowdata.getAfterColumnsList();
                            Map<String, Object> dataMap = transforListToMap(afterColumnsList);
                            if (row.getEventType() == CanalEntry.EventType.INSERT) {
                                //具体业务操作(新增)
                                System.out.println(dataMap);
                            } else if (row.getEventType() == CanalEntry.EventType.UPDATE) {
                                //具体业务操作(修改)
                                System.out.println(dataMap);
                            } else if (row.getEventType() == CanalEntry.EventType.DELETE) {
                                List<CanalEntry.Column> beforeColumnsList = rowdata.getBeforeColumnsList();
                                for (CanalEntry.Column column : beforeColumnsList) {
                                    if ("id".equals(column.getName())) {
                                        //具体业务操作(删除)
                                        System.out.println("删除的id：" + column.getValue());
                                    }
                                }
                            } else {
                                System.out.println("其他操作类型不做处理");
                            }

                        }

                    } catch (InvalidProtocolBufferException e) {
                        e.printStackTrace();
                    }
                }
                //确认消息
                connector.ack(batchId);
            }


        }
    }

    public static Map<String, Object> transforListToMap(List<CanalEntry.Column> afterColumnsList) {
        Map map = new HashMap();
        if (afterColumnsList != null && afterColumnsList.size() > 0) {
            for (CanalEntry.Column column : afterColumnsList) {
                map.put(column.getName(), column.getValue());
            }
        }
        return map;
    }


}