Merge branch 'master' into pyf2

This commit is contained in:
Payon
2018-08-07 19:29:27 +08:00
committed by GitHub
168 changed files with 19475 additions and 11726 deletions

View File

@@ -0,0 +1,89 @@
package com.alibaba.otter.canal.parse;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.sink.exception.CanalSinkException;
public class MysqlBinlogDumpPerformanceTest {
public static void main(String args[]) {
final MysqlEventParser controller = new MysqlEventParser();
final EntryPosition startPosition = new EntryPosition("mysql-bin.001699", 120L, 100L);
controller.setConnectionCharset(Charset.forName("UTF-8"));
controller.setSlaveId(3344L);
controller.setDetectingEnable(false);
controller.setFilterQueryDml(true);
controller.setMasterInfo(new AuthenticationInfo(new InetSocketAddress("127.0.0.1", 3328), "root", "hello"));
controller.setMasterPosition(startPosition);
controller.setEnableTsdb(false);
controller.setDestination("example");
controller.setTsdbSpringXml("classpath:spring/tsdb/h2-tsdb.xml");
// controller.setEventFilter(new AviaterRegexFilter("test\\..*"));
// controller.setEventBlackFilter(new
// AviaterRegexFilter("canal_tsdb\\..*"));
controller.setParallel(true);
controller.setParallelBufferSize(256);
controller.setParallelThreadSize(16);
controller.setIsGTIDMode(false);
final AtomicLong sum = new AtomicLong(0);
final AtomicLong last = new AtomicLong(0);
final AtomicLong start = new AtomicLong(System.currentTimeMillis());
final AtomicLong end = new AtomicLong(0);
controller.setEventSink(new AbstractCanalEventSinkTest<List<CanalEntry.Entry>>() {
public boolean sink(List<CanalEntry.Entry> entrys, InetSocketAddress remoteAddress, String destination)
throws CanalSinkException,
InterruptedException {
sum.addAndGet(entrys.size());
long current = sum.get();
if (current - last.get() >= 100000) {
end.set(System.currentTimeMillis());
long tps = ((current - last.get()) * 1000) / (end.get() - start.get());
System.out.println(" total : " + sum + " , cost : " + (end.get() - start.get()) + " , tps : " + tps);
last.set(current);
start.set(end.get());
}
return true;
}
});
controller.setLogPositionManager(new AbstractLogPositionManager() {
@Override
public LogPosition getLatestIndexBy(String destination) {
return null;
}
@Override
public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
}
});
controller.start();
try {
Thread.sleep(100 * 1000 * 1000L);
} catch (InterruptedException e) {
}
controller.stop();
}
public static abstract class AbstractCanalEventSinkTest<T> extends AbstractCanalLifeCycle implements CanalEventSink<T> {
public void interrupt() {
}
}
}

View File

@@ -0,0 +1,82 @@
package com.alibaba.otter.canal.parse;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor;
import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher;
import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
public class MysqlBinlogEventPerformanceTest {
protected static Charset charset = Charset.forName("utf-8");
public static void main(String args[]) {
DirectLogFetcher fetcher = new DirectLogFetcher();
try {
MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "root", "hello");
connector.connect();
updateSettings(connector);
sendBinlogDump(connector, "mysql-bin.000006", 120L, 3);
fetcher.start(connector.getChannel());
LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
LogContext context = new LogContext();
AtomicLong sum = new AtomicLong(0);
long start = System.currentTimeMillis();
long last = 0;
long end = 0;
while (fetcher.fetch()) {
decoder.decode(fetcher, context);
sum.incrementAndGet();
long current = sum.get();
if (current - last >= 100000) {
end = System.currentTimeMillis();
long tps = ((current - last) * 1000) / (end - start);
System.out.println(" total : " + sum + " , cost : " + (end - start) + " , tps : " + tps);
last = current;
start = end;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
fetcher.close();
} catch (IOException e) {
}
}
}
private static void sendBinlogDump(MysqlConnector connector, String binlogfilename, Long binlogPosition, int slaveId)
throws IOException {
BinlogDumpCommandPacket binlogDumpCmd = new BinlogDumpCommandPacket();
binlogDumpCmd.binlogFileName = binlogfilename;
binlogDumpCmd.binlogPosition = binlogPosition;
binlogDumpCmd.slaveServerId = slaveId;
byte[] cmdBody = binlogDumpCmd.toBytes();
HeaderPacket binlogDumpHeader = new HeaderPacket();
binlogDumpHeader.setPacketBodyLength(cmdBody.length);
binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
}
private static void updateSettings(MysqlConnector connector) throws IOException {
update("set @master_binlog_checksum= '@@global.binlog_checksum'", connector);
update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'", connector);
}
public static void update(String cmd, MysqlConnector connector) throws IOException {
MysqlUpdateExecutor exector = new MysqlUpdateExecutor(connector);
exector.update(cmd);
}
}

View File

@@ -0,0 +1,192 @@
package com.alibaba.otter.canal.parse;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.BitSet;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor;
import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer;
import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent;
import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo;
import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
public class MysqlBinlogParsePerformanceTest {
protected static Charset charset = Charset.forName("utf-8");
public static void main(String args[]) {
DirectLogFetcher fetcher = new DirectLogFetcher();
try {
MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "root", "hello");
connector.connect();
updateSettings(connector);
sendBinlogDump(connector, "mysql-bin.000006", 120L, 3);
fetcher.start(connector.getChannel());
final BlockingQueue<LogBuffer> buffer = new ArrayBlockingQueue<LogBuffer>(1024);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
try {
consumer(buffer);
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
thread.start();
while (fetcher.fetch()) {
buffer.put(fetcher.duplicate());
fetcher.consume(fetcher.limit());
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
fetcher.close();
} catch (IOException e) {
}
}
}
public static void consumer(BlockingQueue<LogBuffer> buffer) throws IOException, InterruptedException {
LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
LogContext context = new LogContext();
AtomicLong sum = new AtomicLong(0);
long start = System.currentTimeMillis();
long last = 0;
long end = 0;
while (true) {
LogEvent event = null;
event = decoder.decode(buffer.take(), context);
int eventType = event.getHeader().getType();
switch (eventType) {
case LogEvent.ROTATE_EVENT:
break;
case LogEvent.WRITE_ROWS_EVENT_V1:
case LogEvent.WRITE_ROWS_EVENT:
parseRowsEvent((WriteRowsLogEvent) event, sum);
break;
case LogEvent.UPDATE_ROWS_EVENT_V1:
case LogEvent.UPDATE_ROWS_EVENT:
parseRowsEvent((UpdateRowsLogEvent) event, sum);
break;
case LogEvent.DELETE_ROWS_EVENT_V1:
case LogEvent.DELETE_ROWS_EVENT:
parseRowsEvent((DeleteRowsLogEvent) event, sum);
break;
case LogEvent.XID_EVENT:
sum.incrementAndGet();
break;
case LogEvent.QUERY_EVENT:
sum.incrementAndGet();
break;
default:
break;
}
long current = sum.get();
if (current - last >= 100000) {
end = System.currentTimeMillis();
long tps = ((current - last) * 1000) / (end - start);
System.out.println(" total : " + sum + " , cost : " + (end - start) + " , tps : " + tps);
last = current;
start = end;
}
}
}
private static void sendBinlogDump(MysqlConnector connector, String binlogfilename, Long binlogPosition, int slaveId)
throws IOException {
BinlogDumpCommandPacket binlogDumpCmd = new BinlogDumpCommandPacket();
binlogDumpCmd.binlogFileName = binlogfilename;
binlogDumpCmd.binlogPosition = binlogPosition;
binlogDumpCmd.slaveServerId = slaveId;
byte[] cmdBody = binlogDumpCmd.toBytes();
HeaderPacket binlogDumpHeader = new HeaderPacket();
binlogDumpHeader.setPacketBodyLength(cmdBody.length);
binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
}
private static void updateSettings(MysqlConnector connector) throws IOException {
update("set @master_binlog_checksum= '@@global.binlog_checksum'", connector);
update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'", connector);
}
public static void update(String cmd, MysqlConnector connector) throws IOException {
MysqlUpdateExecutor exector = new MysqlUpdateExecutor(connector);
exector.update(cmd);
}
public static void parseRowsEvent(RowsLogEvent event, AtomicLong sum) {
try {
RowsLogBuffer buffer = event.getRowsBuf(charset.name());
BitSet columns = event.getColumns();
BitSet changeColumns = event.getChangeColumns();
while (buffer.nextOneRow(columns)) {
int type = event.getHeader().getType();
if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
parseOneRow(event, buffer, columns, true);
} else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type) {
parseOneRow(event, buffer, columns, false);
} else {
parseOneRow(event, buffer, columns, false);
if (!buffer.nextOneRow(changeColumns)) {
break;
}
parseOneRow(event, buffer, changeColumns, true);
}
sum.incrementAndGet();
}
} catch (Exception e) {
throw new RuntimeException("parse row data failed.", e);
}
}
public static void parseOneRow(RowsLogEvent event, RowsLogBuffer buffer, BitSet cols, boolean isAfter)
throws UnsupportedEncodingException {
TableMapLogEvent map = event.getTable();
if (map == null) {
throw new RuntimeException("not found TableMap with tid=" + event.getTableId());
}
final int columnCnt = map.getColumnCnt();
final ColumnInfo[] columnInfo = map.getColumnInfo();
for (int i = 0; i < columnCnt; i++) {
if (!cols.get(i)) {
continue;
}
ColumnInfo info = columnInfo[i];
buffer.nextValue(info.type, info.meta);
if (buffer.isNull()) {
} else {
buffer.getValue();
}
}
}
}

View File

@@ -7,7 +7,6 @@ import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.position.Position;
import com.alibaba.otter.canal.store.CanalEventStore;
@@ -18,7 +17,7 @@ import com.alibaba.otter.canal.store.model.Events;
public class DummyEventStore implements CanalEventStore<Event> {
private static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
private static final String messgae = "{0} [{1}:{2}:{3}] {4} {5}.{6}";
private static final String messgae = "{0} [{1}:{2}:{3}]";
public void ack(Position position) throws CanalStoreException {
@@ -82,46 +81,40 @@ public class DummyEventStore implements CanalEventStore<Event> {
public void put(List<Event> datas) throws InterruptedException, CanalStoreException {
for (Event data : datas) {
CanalEntry.Header header = data.getEntry().getHeader();
Date date = new Date(header.getExecuteTime());
Date date = new Date(data.getExecuteTime());
SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
if (data.getEntry().getEntryType() == EntryType.TRANSACTIONBEGIN
|| data.getEntry().getEntryType() == EntryType.TRANSACTIONEND) {
if (data.getEntryType() == EntryType.TRANSACTIONBEGIN || data.getEntryType() == EntryType.TRANSACTIONEND) {
// System.out.println(MessageFormat.format(messgae, new Object[]
// { Thread.currentThread().getName(),
// header.getLogfilename(), header.getLogfileoffset(),
// format.format(date),
// data.getEntry().getEntryType(), "" }));
System.out.println(data.getEntry().getEntryType());
System.out.println(data.getEntryType());
} else {
System.out.println(MessageFormat.format(messgae,
new Object[] { Thread.currentThread().getName(), header.getLogfileName(),
String.valueOf(header.getLogfileOffset()), format.format(date), header.getEventType(),
header.getSchemaName(), header.getTableName() }));
new Object[] { Thread.currentThread().getName(), data.getJournalName(),
String.valueOf(data.getPosition()), format.format(date) }));
}
}
}
public boolean put(List<Event> datas, long timeout, TimeUnit unit) throws InterruptedException, CanalStoreException {
for (Event data : datas) {
CanalEntry.Header header = data.getEntry().getHeader();
Date date = new Date(header.getExecuteTime());
Date date = new Date(data.getExecuteTime());
SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
if (data.getEntry().getEntryType() == EntryType.TRANSACTIONBEGIN
|| data.getEntry().getEntryType() == EntryType.TRANSACTIONEND) {
if (data.getEntryType() == EntryType.TRANSACTIONBEGIN || data.getEntryType() == EntryType.TRANSACTIONEND) {
// System.out.println(MessageFormat.format(messgae, new Object[]
// { Thread.currentThread().getName(),
// header.getLogfilename(), header.getLogfileoffset(),
// format.format(date),
// data.getEntry().getEntryType(), "" }));
System.out.println(data.getEntry().getEntryType());
System.out.println(data.getEntryType());
} else {
System.out.println(MessageFormat.format(messgae,
new Object[] { Thread.currentThread().getName(), header.getLogfileName(),
String.valueOf(header.getLogfileOffset()), format.format(date), header.getEventType(),
header.getSchemaName(), header.getTableName() }));
new Object[] { Thread.currentThread().getName(), data.getJournalName(),
String.valueOf(data.getPosition()), format.format(date) }));
}
}
return true;
@@ -131,23 +124,20 @@ public class DummyEventStore implements CanalEventStore<Event> {
System.out.println("\n>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
for (Event data : datas) {
CanalEntry.Header header = data.getEntry().getHeader();
Date date = new Date(header.getExecuteTime());
Date date = new Date(data.getExecuteTime());
SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
if (data.getEntry().getEntryType() == EntryType.TRANSACTIONBEGIN
|| data.getEntry().getEntryType() == EntryType.TRANSACTIONEND) {
if (data.getEntryType() == EntryType.TRANSACTIONBEGIN || data.getEntryType() == EntryType.TRANSACTIONEND) {
// System.out.println(MessageFormat.format(messgae, new Object[]
// { Thread.currentThread().getName(),
// header.getLogfilename(), header.getLogfileoffset(),
// format.format(date),
// data.getEntry().getEntryType(), "" }));
System.out.println(data.getEntry().getEntryType());
System.out.println(data.getEntryType());
} else {
System.out.println(MessageFormat.format(messgae,
new Object[] { Thread.currentThread().getName(), header.getLogfileName(),
String.valueOf(header.getLogfileOffset()), format.format(date), header.getEventType(),
header.getSchemaName(), header.getTableName() }));
new Object[] { Thread.currentThread().getName(), data.getJournalName(),
String.valueOf(data.getPosition()), format.format(date) }));
}
}

View File

@@ -41,6 +41,9 @@ public class MysqlDumpTest {
controller.setTsdbSpringXml("classpath:tsdb/h2-tsdb.xml");
controller.setEventFilter(new AviaterRegexFilter("test\\..*"));
controller.setEventBlackFilter(new AviaterRegexFilter("canal_tsdb\\..*"));
controller.setParallel(true);
controller.setParallelBufferSize(256);
controller.setParallelThreadSize(2);
controller.setIsGTIDMode(false);
controller.setEventSink(new AbstractCanalEventSinkTest<List<Entry>>() {

View File

@@ -0,0 +1,30 @@
package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
import java.io.FileNotFoundException;
import java.io.IOException;
import org.junit.Test;
import com.alibaba.fastsql.sql.repository.SchemaObject;
import com.alibaba.fastsql.sql.repository.SchemaRepository;
import com.alibaba.fastsql.util.JdbcConstants;
/**
* @author agapple 2018年6月7日 下午5:36:13
* @since 3.1.9
*/
public class FastsqlSchemaTest {
@Test
public void testSimple() throws FileNotFoundException, IOException {
SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL);
String sql = "create table yushitai_test.card_record ( id bigint auto_increment) auto_increment=256 "
+ "; alter table yushitai_test.card_record add column customization_id bigint unsigned NOT NULL COMMENT 'TEST' ;"
+ "; rename table yushitai_test.card_record to yushitai_test._card_record_del;";
repository.console(sql);
repository.setDefaultSchema("yushitai_test");
SchemaObject table = repository.findTable("_card_record_del");
System.out.println(table.getStatement().toString());
}
}

View File

@@ -0,0 +1,40 @@
package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
import java.io.File;
import java.io.FileInputStream;
import java.net.URL;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.alibaba.otter.canal.parse.inbound.TableMeta;
/**
* @author agapple 2017年8月1日 下午7:15:54
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/tsdb/mysql-tsdb.xml" })
public class MemoryTableMeta_DDL_Test {
@Test
public void test1() throws Throwable {
MemoryTableMeta memoryTableMeta = new MemoryTableMeta();
URL url = Thread.currentThread().getContextClassLoader().getResource("dummy.txt");
File dummyFile = new File(url.getFile());
File create = new File(dummyFile.getParent() + "/ddl", "ddl_test1.sql");
String sql = StringUtils.join(IOUtils.readLines(new FileInputStream(create)), "\n");
memoryTableMeta.apply(null, "test", sql, null);
TableMeta meta = memoryTableMeta.find("yushitai_test", "card_record");
System.out.println(meta);
Assert.assertNotNull(meta.getFieldMetaByName("customization_id"));
meta = memoryTableMeta.find("yushitai_test", "_card_record_gho");
Assert.assertNull(meta);
}
}

View File

@@ -0,0 +1,21 @@
create table yushitai_test.card_record (
id bigint auto_increment,
last_update timestamp not null DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
hint varchar(64) charset ascii not null,
value varchar(255) charset ascii not null,
primary key(id),
unique key hint_uidx(hint)
) auto_increment=256 ;
DROP TABLE IF EXISTS _card_record_gho /* generated by server */ ;
DROP TABLE IF EXISTS _card_record_del /* generated by server */ ;
create /* gh-ost */ table yushitai_test._card_record_gho like yushitai_test.card_record ;
alter /* gh-ost */ table yushitai_test._card_record_gho add column customization_id bigint unsigned NOT NULL COMMENT 'TEST' ;
create /* gh-ost */ table yushitai_test._card_record_del (
id int auto_increment primary key
) engine=InnoDB comment='ghost-cut-over-sentry' ;
DROP TABLE IF EXISTS _card_record_del /* generated by server */ ;
rename /* gh-ost */ table yushitai_test.card_record to yushitai_test._card_record_del, yushitai_test._card_record_gho to yushitai_test.card_record;