public class CanalClientApi {
public static void main(String args[]) {
// 创建canal连接对象
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("xxx",
11111), "example", "canal", "canal");
System.out.println(connector.toString());
try {
//连接
connector.connect();
//订阅所有数据库和表
connector.subscribe(".*\..*");
connector.rollback();
while (true) {
// 获取指定数量的数据
Message message = connector.getWithoutAck(1000);
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
//没有数据，就休眠1秒
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
} else {
//有数据就打印
printEntry(message.getEntries());
}
// 提交确认
connector.ack(batchId);
}
} finally {
connector.disconnect();
}
}
private static void printEntry(List entrys) {
for (Entry entry : entrys) {
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN ||
entry.getEntryType() == EntryType.TRANSACTIONEND) {
continue;
}
RowChange rowChage = null;
try {
rowChage = RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR parse data:" + entry.toString(),e);
}
EventType eventType = rowChage.getEventType();
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));
for (RowData rowData : rowChage.getRowDatasList()) {
//判断增删改操作
if (eventType == EventType.DELETE) {
printColumn(rowData.getBeforeColumnsList());
} else if (eventType == 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 columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}
注意：在配置过程中一直报错，信息为
Caused by: java.io.IOException: ErrorPacket [errorNumber=1146, fieldCount=-1, message=Table 'mysql.BASE TABLE' doesn't exist, sqlState=42S02, sqlStateMarker=#]
with command: show create table `mysql`.`columns_priv`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`db`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`engine_cost`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`event`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`func`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`general_log`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`gtid_executed`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`help_category`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`help_keyword`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`help_relation`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`help_topic`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`innodb_index_stats`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`innodb_table_stats`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`ndb_binlog_index`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`plugin`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`proc`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`procs_priv`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`proxies_priv`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`server_cost`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`servers`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`slow_log`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`tables_priv`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`time_zone`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`time_zone_leap_second`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`time_zone_name`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`time_zone_transition`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`time_zone_transition_type`;show create table `mysql`.`BASE TABLE`;show create table `mysql`.`user`;show create table `mysql`.`BASE TABLE`;
at com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor.queryMulti(MysqlQueryExecutor.java:109)
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.queryMulti(MysqlConnection.java:111)
at com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta.dumpTableMeta(DatabaseTableMeta.java:233)
at com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta.rollback(DatabaseTableMeta.java:174)
at com.alibaba.otter.canal.parse.inbound.mysql.AbstractMysqlEventParser.processTableMeta(AbstractMysqlEventParser.java:142)
at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$1.run(AbstractEventParser.java:197
1.1.4，instance启用，日志循环create hive表的意义 · Issue #2499 · alibaba/canal · GitHub
