win10 安装使用canal1.1.5
环境 ：win10，mysql5.7，canal1.1.5最新
1. 修改mysql配置，位置（D:\soft\mysql-5.7.30-winx64\my.ini）
[mysqld] 服务端配置下面添加三行（其他配置勿动）：
server_id=1
log-bin=master
binlog_format=row
注意：mysql实例id，不能和canal的slaveId重复
或者直接运行如下指令：
mysql> set global binlog_format=ROW;
mysql> set global binlog_row_image=FULL;
配置完后重启mysql 服务：net stop mysql net start mysql
canal需要建库、建用户和赋权：
CREATE DATABASE `canal_manager`;
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
-- 注意权限问题，否则在后续的一些操作中会提示没有权限
GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
2. 下载安装canal服务端
2.1. 解压第一步下载的canal.deployer-1.1.5-SNAPSHOT.tar.gz
https://github.com/alibaba/canal/releases/
安装canal，单机只需安装一个服务（canal.deployer）:
2.2. 修改conf/canal.properties配置文件
注册地址为本机IP(canal.register.ip =本机IP)，还要配置输出到tcp（java接口作为一个客户端可收到信息）或kafka
canal.port = 11111 #对外提供的socket接口。如端口不冲突，使用默认即可
canal.serverMode = tcp# tcp对应接口；选择kafka的推送模式，发送kafka消息
另外配置目的地，样例是 example，conf下有这个文件夹。
canal.destinations = example
2.3. 修改conf/example/instance.properties配置文件
#配置数据库信息
canal.instance.master.address=127.0.0.1:3306
#show master status 可查看日志文件及其位置，对应这个参数canal.instance.master.journal.name=master.000004
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
2.4. 启动canal服务端
./bin/startup.bat
3. 启动java客户端接受信息
一般用kafka接受信息，这里使用java springboot接收信息，启动主方法即可，修改数据库信息可以看到消息。
package org.demo.springbootshirocasdemo.component;
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 lombok.SneakyThrows;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import java.net.InetSocketAddress;
import java.util.List;
public class CanalClient {
private static final int BATCH_SIZE = 1;
private static final String HOST_NAME ="127.0.0.1";
private static final int PORT =11111;
private static final String DESTINATION ="example";
private static final String USERNAME = "root";
private static final String PASSWORD = "root";
public static void main(String[] args) {
try {
startCanal();
} catch (Exception e) {
e.printStackTrace();
}
}
public static void startCanal() throws Exception {
CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress(HOST_NAME,PORT), DESTINATION, USERNAME, PASSWORD);
try {
connector.connect();
//订阅数据库表,全部表
connector.subscribe(".*\\..*");
//回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
connector.rollback();
while (true) {
// 获取指定数量的数据
Message message = connector.getWithoutAck(BATCH_SIZE);
//获取批量ID
long batchId = message.getId();
//获取批量的数量
int size = message.getEntries().size();
//如果没有数据
if (batchId == -1 || size == 0) {
try {
//线程休眠2秒
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
//如果有数据,处理数据
printEntry(message.getEntries());
}
//进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
connector.ack(batchId);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
connector.disconnect();
}
}
/**
* 打印canal server解析binlog获得的实体类信息
*/
private static void printEntry(List<CanalEntry.Entry> entrys) {
for (CanalEntry.Entry entry : entrys) {
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
//开启/关闭事务的实体类型，跳过
continue;
}
//RowChange对象，包含了一行数据变化的所有特征
//比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
CanalEntry.RowChange rowChage;
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);
}
//获取操作类型：insert/update/delete类型
CanalEntry.EventType eventType = rowChage.getEventType();
//打印Header信息
System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s",
entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
eventType));
//判断是否是DDL语句
if (rowChage.getIsDdl()) {
System.out.println("================》;isDdl: true,sql:" + rowChage.getSql());
}
//获取RowChange对象里的每一行数据，打印出来
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("------->; before");
printColumn(rowData.getBeforeColumnsList());
//变更后的数据
System.out.println("------->; after");
printColumn(rowData.getAfterColumnsList());
}
}
}
}
private static void printColumn(List<CanalEntry.Column> columns) {
for (CanalEntry.Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}
4. 可能收到的错误
乱码
vim ../conf/canal.properties
#传输格式：json
canal.mq.flatMessage = true
