chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

= cold cleanup

Browse Source 
* using lambda if possible (we are already building in Java8)
  * using diamond when new generic instance
  * remove used improt
This commit is contained in:
zavakid
2020-09-17 10:48:15 +08:00
parent df60eda679
commit e7658879bc
133 changed files with 990 additions and 1558 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
View File

@@ -80,14 +80,8 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
protected Thread parseThread = null;
protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
public void uncaughtException(Thread t,
Throwable e) {
logger.error("parse events has an error",
e);
}
};
protected Thread.UncaughtExceptionHandler handler = (t, e) -> logger.error("parse events has an error",
e);
protected EventTransactionBuffer transactionBuffer;
protected int transactionSize = 1024;
@@ -134,22 +128,19 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
public AbstractEventParser(){
// 初始化一下
transactionBuffer = new EventTransactionBuffer(new TransactionFlushCallback() {
transactionBuffer = new EventTransactionBuffer(transaction -> {
boolean successed = consumeTheEventAndProfilingIfNecessary(transaction);
if (!running) {
return;
}
public void flush(List<CanalEntry.Entry> transaction) throws InterruptedException {
boolean successed = consumeTheEventAndProfilingIfNecessary(transaction);
if (!running) {
return;
}
if (!successed) {
throw new CanalParseException("consume failed!");
}
if (!successed) {
throw new CanalParseException("consume failed!");
}
LogPosition position = buildLastTransactionPosition(transaction);
if (position != null) { // 可能position为空
logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position);
}
LogPosition position = buildLastTransactionPosition(transaction);
if (position != null) { // 可能position为空
logPositionManager.persistLogPosition(AbstractEventParser.this.destination, position);
}
});
}
@@ -542,7 +533,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
*/
private Map<String, List<String>> parseFieldFilterMap(String config) {
Map<String, List<String>> map = new HashMap<String, List<String>>();
Map<String, List<String>> map = new HashMap<>();
if (StringUtils.isNotBlank(config)) {
for (String filter : config.split(",")) {

2
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/EventTransactionBuffer.java
View File

@@ -115,7 +115,7 @@ public class EventTransactionBuffer extends AbstractCanalLifeCycle {
long end = this.putSequence.get();
if (start <= end) {
List<CanalEntry.Entry> transaction = new ArrayList<CanalEntry.Entry>();
List<CanalEntry.Entry> transaction = new ArrayList<>();
for (long next = start; next <= end; next++) {
transaction.add(this.entries[getIndex(next)]);
}

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java
View File

@@ -23,7 +23,7 @@ public class TableMeta {
private String schema;
private String table;
private List<FieldMeta> fields = new ArrayList<TableMeta.FieldMeta>();
private List<FieldMeta> fields = new ArrayList<>();
private String ddl; // 表结构的DDL语句
public TableMeta(){
@@ -75,7 +75,7 @@ public class TableMeta {
}
public List<FieldMeta> getPrimaryFields() {
List<FieldMeta> primarys = new ArrayList<TableMeta.FieldMeta>();
List<FieldMeta> primarys = new ArrayList<>();
for (FieldMeta meta : fields) {
if (meta.isKey()) {
primarys.add(meta);

2
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventParser.java
View File

@@ -14,7 +14,7 @@ import com.alibaba.otter.canal.parse.CanalEventParser;
*/
public class GroupEventParser extends AbstractCanalLifeCycle implements CanalEventParser {
private List<CanalEventParser> eventParsers = new ArrayList<CanalEventParser>();
private List<CanalEventParser> eventParsers = new ArrayList<>();
public void start() {
super.start();

2
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
View File

@@ -714,7 +714,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
}
int i = 0;
Map<String, String> maps = new HashMap<String, String>(names.size(), 1f);
Map<String, String> maps = new HashMap<>(names.size(), 1f);
for (FieldPacket name : names) {
maps.put(name.getName(), fields.get(i));
i++;

12
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
View File

@@ -98,9 +98,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
ExceptionHandler exceptionHandler = new SimpleFatalExceptionHandler();
// stage 2
this.logContext = new LogContext();
simpleParserStage = new BatchEventProcessor<MessageEvent>(disruptorMsgBuffer,
sequenceBarrier,
new SimpleParserStage(logContext));
simpleParserStage = new BatchEventProcessor<>(disruptorMsgBuffer,
sequenceBarrier,
new SimpleParserStage(logContext));
simpleParserStage.setExceptionHandler(exceptionHandler);
disruptorMsgBuffer.addGatingSequences(simpleParserStage.getSequence());
@@ -119,9 +119,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
// stage 4
SequenceBarrier sinkSequenceBarrier = disruptorMsgBuffer.newBarrier(sequence);
sinkStoreStage = new BatchEventProcessor<MessageEvent>(disruptorMsgBuffer,
sinkSequenceBarrier,
new SinkStoreStage());
sinkStoreStage = new BatchEventProcessor<>(disruptorMsgBuffer,
sinkSequenceBarrier,
new SinkStoreStage());
sinkStoreStage.setExceptionHandler(exceptionHandler);
disruptorMsgBuffer.addGatingSequences(sinkStoreStage.getSequence());

8
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
View File

@@ -88,8 +88,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
private volatile AviaterRegexFilter nameFilter; // 运行时引用可能会有变化，比如规则发生变化时
private volatile AviaterRegexFilter nameBlackFilter;
private Map<String, List<String>> fieldFilterMap = new HashMap<String, List<String>>();
private Map<String, List<String>> fieldBlackFilterMap = new HashMap<String, List<String>>();
private Map<String, List<String>> fieldFilterMap = new HashMap<>();
private Map<String, List<String>> fieldBlackFilterMap = new HashMap<>();
private TableMetaCache tableMetaCache;
private Charset charset = Charset.defaultCharset();
@@ -1040,7 +1040,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
if (fieldFilterMap != null) {
this.fieldFilterMap = fieldFilterMap;
} else {
this.fieldFilterMap = new HashMap<String, List<String>>();
this.fieldFilterMap = new HashMap<>();
}
for (Map.Entry<String, List<String>> entry : this.fieldFilterMap.entrySet()) {
@@ -1052,7 +1052,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
if (fieldBlackFilterMap != null) {
this.fieldBlackFilterMap = fieldBlackFilterMap;
} else {
this.fieldBlackFilterMap = new HashMap<String, List<String>>();
this.fieldBlackFilterMap = new HashMap<>();
}
for (Map.Entry<String, List<String>> entry : this.fieldBlackFilterMap.entrySet()) {

6
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java
View File

@@ -105,7 +105,7 @@ public class TableMetaCache {
TableMeta tableMeta = memoryTableMeta.find(schema, table);
return tableMeta.getFields();
} else {
return new ArrayList<FieldMeta>();
return new ArrayList<>();
}
}
@@ -113,7 +113,7 @@ public class TableMetaCache {
* 处理desc table的结果
*/
public static List<FieldMeta> parseTableMetaByDesc(ResultSetPacket packet) {
Map<String, Integer> nameMaps = new HashMap<String, Integer>(6, 1f);
Map<String, Integer> nameMaps = new HashMap<>(6, 1f);
int index = 0;
for (FieldPacket fieldPacket : packet.getFieldDescriptors()) {
nameMaps.put(fieldPacket.getOriginalName(), index++);
@@ -121,7 +121,7 @@ public class TableMetaCache {
int size = packet.getFieldDescriptors().size();
int count = packet.getFieldValues().size() / packet.getFieldDescriptors().size();
List<FieldMeta> result = new ArrayList<FieldMeta>();
List<FieldMeta> result = new ArrayList<>();
for (int i = 0; i < count; i++) {
FieldMeta meta = new FieldMeta();
// 做一个优化，使用String.intern()，共享String对象，减少内存使用

2
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java
View File

@@ -56,7 +56,7 @@ public class DruidDdlParser {
return Arrays.asList(ddlResult);
}
List<DdlResult> ddlResults = new ArrayList<DdlResult>();
List<DdlResult> ddlResults = new ArrayList<>();
for (SQLStatement statement : stmtList) {
if (statement instanceof SQLCreateTableStatement) {
DdlResult ddlResult = new DdlResult();

14
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java
View File

@@ -26,7 +26,7 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException;
public class BinLogFileQueue {
private String baseName = "mysql-bin.";
private List<File> binlogs = new ArrayList<File>();
private List<File> binlogs = new ArrayList<>();
private File directory;
private ReentrantLock lock = new ReentrantLock();
private Condition nextCondition = lock.newCondition();
@@ -177,7 +177,7 @@ public class BinLogFileQueue {
* 获取当前所有binlog文件
*/
public List<File> currentBinlogs() {
return new ArrayList<File>(binlogs);
return new ArrayList<>(binlogs);
}
public void destory() {
@@ -216,7 +216,7 @@ public class BinLogFileQueue {
}
private List<File> listBinlogFiles() {
List<File> files = new ArrayList<File>();
List<File> files = new ArrayList<>();
files.addAll(FileUtils.listFiles(directory, new IOFileFilter() {
public boolean accept(File file) {
@@ -230,13 +230,7 @@ public class BinLogFileQueue {
}
}, null));
// 排一下序列
Collections.sort(files, new Comparator<File>() {
public int compare(File o1, File o2) {
return o1.getName().compareTo(o2.getName());
}
});
Collections.sort(files, Comparator.comparing(File::getName));
return files;
}

22
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java
View File

@@ -53,8 +53,8 @@ public class BinlogDownloadQueue {
private static final Logger logger = LoggerFactory.getLogger(BinlogDownloadQueue.class);
private static final int TIMEOUT = 10000;
private LinkedBlockingQueue<BinlogFile> downloadQueue = new LinkedBlockingQueue<BinlogFile>();
private LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<Runnable>();
private LinkedBlockingQueue<BinlogFile> downloadQueue = new LinkedBlockingQueue<>();
private LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();
private LinkedList<BinlogFile> binlogList;
private final int batchFileSize;
private Thread downloadThread;
@@ -78,13 +78,7 @@ public class BinlogDownloadQueue {
String fileName = StringUtils.substringBetween(binlog.getDownloadLink(), "mysql-bin.", "?");
binlog.setFileName(fileName);
}
Collections.sort(this.binlogList, new Comparator<BinlogFile>() {
@Override
public int compare(BinlogFile o1, BinlogFile o2) {
return o1.getFileName().compareTo(o2.getFileName());
}
});
this.binlogList.sort(Comparator.comparing(BinlogFile::getFileName));
}
public void cleanDir() throws IOException {
@@ -187,13 +181,9 @@ public class BinlogDownloadQueue {
builder.setMaxConnPerRoute(50);
builder.setMaxConnTotal(100);
// 创建支持忽略证书的https
final SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new TrustStrategy() {
@Override
public boolean isTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {
return true;
}
}).build();
final SSLContext sslContext = new SSLContextBuilder()
.loadTrustMaterial(null, (x509Certificates, s) -> true)
.build();
httpClient = HttpClientBuilder.create()
.setSSLContext(sslContext)

20
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java
View File

@@ -140,13 +140,9 @@ public class HttpHelper {
CloseableHttpResponse response = null;
try {
// 创建支持忽略证书的https
final SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new TrustStrategy() {
@Override
public boolean isTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {
return true;
}
}).build();
final SSLContext sslContext = new SSLContextBuilder()
.loadTrustMaterial(null, (x509Certificates, s) -> true)
.build();
CloseableHttpClient httpClient = HttpClientBuilder.create()
.setSSLContext(sslContext)
@@ -205,13 +201,9 @@ public class HttpHelper {
CloseableHttpResponse response = null;
try {
// 创建支持忽略证书的https
final SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, new TrustStrategy() {
@Override
public boolean isTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {
return true;
}
}).build();
final SSLContext sslContext = new SSLContextBuilder()
.loadTrustMaterial(null, (x509Certificates, s) -> true)
.build();
CloseableHttpClient httpClient = HttpClientBuilder.create()
.setSSLContext(sslContext)

85
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
View File

@@ -30,16 +30,12 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
private int batchFileSize = 4; // 最多下载的binlog文件数量
private RdsLocalBinlogEventParser rdsLocalBinlogEventParser = null;
private ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
private ExecutorService executorService = Executors.newSingleThreadExecutor(r -> {
Thread t = new Thread(r,
"rds-binlog-daemon-thread");
t.setDaemon(true);
return t;
}
});
});
@Override
public void start() {
@@ -79,32 +75,17 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
rdsLocalBinlogEventParser.setParallel(this.parallel);
rdsLocalBinlogEventParser.setParallelBufferSize(this.parallelBufferSize);
rdsLocalBinlogEventParser.setParallelThreadSize(this.parallelThreadSize);
rdsLocalBinlogEventParser.setFinishListener(new RdsLocalBinlogEventParser.ParseFinishListener() {
@Override
public void onFinish() {
executorService.execute(new Runnable() {
@Override
public void run() {
rdsLocalBinlogEventParser.stop();
// empty the dump error count,or will go into local binlog mode again,with error
// position,never get out,fixed by bucketli
RdsBinlogEventParserProxy.this.setDumpErrorCount(0);
RdsBinlogEventParserProxy.this.start();
}
});
}
});
this.setParserExceptionHandler(new ParserExceptionHandler() {
@Override
public void handle(Throwable e) {
handleMysqlParserException(e);
if (targetHandler != null) {
targetHandler.handle(e);
}
rdsLocalBinlogEventParser.setFinishListener(() -> executorService.execute(() -> {
rdsLocalBinlogEventParser.stop();
// empty the dump error count,or will go into local binlog mode again,with error
// position,never get out,fixed by bucketli
RdsBinlogEventParserProxy.this.setDumpErrorCount(0);
RdsBinlogEventParserProxy.this.start();
}));
this.setParserExceptionHandler(e -> {
handleMysqlParserException(e);
if (targetHandler != null) {
targetHandler.handle(e);
}
});
}
@@ -115,29 +96,25 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
private void handleMysqlParserException(Throwable throwable) {
if (throwable instanceof PositionNotFoundException) {
logger.info("remove rds not found position, try download rds binlog!");
executorService.execute(new Runnable() {
executorService.execute(() -> {
try {
logger.info("stop mysql parser!");
RdsBinlogEventParserProxy rdsBinlogEventParserProxy = RdsBinlogEventParserProxy.this;
long serverId = rdsBinlogEventParserProxy.getServerId();
rdsLocalBinlogEventParser.setServerId(serverId);
rdsBinlogEventParserProxy.stop();
} catch (Throwable e) {
logger.info("handle exception failed", e);
}
@Override
public void run() {
try {
logger.info("stop mysql parser!");
RdsBinlogEventParserProxy rdsBinlogEventParserProxy = RdsBinlogEventParserProxy.this;
long serverId = rdsBinlogEventParserProxy.getServerId();
rdsLocalBinlogEventParser.setServerId(serverId);
rdsBinlogEventParserProxy.stop();
} catch (Throwable e) {
logger.info("handle exception failed", e);
}
try {
logger.info("start rds mysql binlog parser!");
rdsLocalBinlogEventParser.start();
} catch (Throwable e) {
logger.info("handle exception failed", e);
rdsLocalBinlogEventParser.stop();
RdsBinlogEventParserProxy rdsBinlogEventParserProxy = RdsBinlogEventParserProxy.this;
rdsBinlogEventParserProxy.start();// 继续重试
}
try {
logger.info("start rds mysql binlog parser!");
rdsLocalBinlogEventParser.start();
} catch (Throwable e) {
logger.info("handle exception failed", e);
rdsLocalBinlogEventParser.stop();
RdsBinlogEventParserProxy rdsBinlogEventParserProxy = RdsBinlogEventParserProxy.this;
rdsBinlogEventParserProxy.start();// 继续重试
}
});
}

18
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java
View File

@@ -82,13 +82,7 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
logger.error("download binlog failed", e);
throw new CanalParseException(e);
}
setParserExceptionHandler(new ParserExceptionHandler() {
@Override
public void handle(Throwable e) {
handleMysqlParserException(e);
}
});
setParserExceptionHandler(this::handleMysqlParserException);
super.start();
}
@@ -105,13 +99,9 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
}
try {
binlogDownloadQueue.execute(new Runnable() {
@Override
public void run() {
RdsLocalBinlogEventParser.super.stop();
RdsLocalBinlogEventParser.super.start();
}
binlogDownloadQueue.execute(() -> {
RdsLocalBinlogEventParser.super.stop();
RdsLocalBinlogEventParser.super.start();
});
} catch (InterruptedException e) {
throw new RuntimeException(e);

10
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/request/AbstractRequest.java
View File

@@ -177,7 +177,7 @@ public abstract class AbstractRequest<T> {
private String makeRequestString(Map<String, String> param) throws Exception {
fillCommonParam(param);
String sign = makeSignature(new TreeMap<String, String>(param));
String sign = makeSignature(new TreeMap<>(param));
StringBuilder builder = new StringBuilder();
for (Map.Entry<String, String> entry : param.entrySet()) {
builder.append(encode(entry.getKey())).append("=").append(encode(entry.getValue())).append("&");
@@ -195,13 +195,7 @@ public abstract class AbstractRequest<T> {
*/
@SuppressWarnings("deprecation")
private final HttpResponse executeHttpRequest(HttpGet getMethod, String host) throws Exception {
SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(null, new TrustStrategy() {
@Override
public boolean isTrusted(X509Certificate[] arg0, String arg1) throws CertificateException {
return true;
}
}).build();
SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(null, (TrustStrategy) (arg0, arg1) -> true).build();
SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(sslContext,
new String[] { "TLSv1" },
null,

60
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
View File

@@ -52,16 +52,12 @@ public class DatabaseTableMeta implements TableMetaTSDB {
private static Logger logger = LoggerFactory.getLogger(DatabaseTableMeta.class);
private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'");
private static Pattern h2Pattern = Pattern.compile("Unique index or primary key violation");
private static ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r,
"[scheduler-table-meta-snapshot]");
thread.setDaemon(true);
return thread;
}
});
private static ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
Thread thread = new Thread(r,
"[scheduler-table-meta-snapshot]");
thread.setDaemon(true);
return thread;
});
private ReadWriteLock lock = new ReentrantReadWriteLock();
private AtomicBoolean initialized = new AtomicBoolean(false);
private String destination;
@@ -69,8 +65,8 @@ public class DatabaseTableMeta implements TableMetaTSDB {
private volatile MysqlConnection connection; // 查询meta信息的链接
private CanalEventFilter filter;
private CanalEventFilter blackFilter;
private Map<String, List<String>> fieldFilterMap = new HashMap<String, List<String>>();
private Map<String, List<String>> fieldBlackFilterMap = new HashMap<String, List<String>>();
private Map<String, List<String>> fieldFilterMap = new HashMap<>();
private Map<String, List<String>> fieldBlackFilterMap = new HashMap<>();
private EntryPosition lastPosition;
private boolean hasNewDdl;
private MetaHistoryDAO metaHistoryDAO;
@@ -91,26 +87,22 @@ public class DatabaseTableMeta implements TableMetaTSDB {
// 24小时生成一份snapshot
if (snapshotInterval > 0) {
scheduleSnapshotFuture = scheduler.scheduleWithFixedDelay(new Runnable() {
scheduleSnapshotFuture = scheduler.scheduleWithFixedDelay(() -> {
boolean applyResult = false;
try {
MDC.put("destination", destination);
applyResult = applySnapshotToDB(lastPosition, false);
} catch (Throwable e) {
logger.error("scheudle applySnapshotToDB faield", e);
}
@Override
public void run() {
boolean applyResult = false;
try {
MDC.put("destination", destination);
applyResult = applySnapshotToDB(lastPosition, false);
} catch (Throwable e) {
logger.error("scheudle applySnapshotToDB faield", e);
}
try {
MDC.put("destination", destination);
if (applyResult) {
snapshotExpire((int) TimeUnit.HOURS.toSeconds(snapshotExpire));
}
} catch (Throwable e) {
logger.error("scheudle snapshotExpire faield", e);
try {
MDC.put("destination", destination);
if (applyResult) {
snapshotExpire((int) TimeUnit.HOURS.toSeconds(snapshotExpire));
}
} catch (Throwable e) {
logger.error("scheudle snapshotExpire faield", e);
}
}, snapshotInterval, snapshotInterval, TimeUnit.HOURS);
}
@@ -199,7 +191,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
private boolean dumpTableMeta(MysqlConnection connection, final CanalEventFilter filter) {
try {
ResultSetPacket packet = connection.query("show databases");
List<String> schemas = new ArrayList<String>();
List<String> schemas = new ArrayList<>();
for (String schema : packet.getFieldValues()) {
schemas.add(schema);
}
@@ -207,7 +199,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
for (String schema : schemas) {
// filter views
packet = connection.query("show full tables from `" + schema + "` where Table_type = 'BASE TABLE'");
List<String> tables = new ArrayList<String>();
List<String> tables = new ArrayList<>();
for (String table : packet.getFieldValues()) {
if ("BASE TABLE".equalsIgnoreCase(table)) {
continue;
@@ -245,7 +237,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
}
private boolean applyHistoryToDB(EntryPosition position, String schema, String ddl, String extra) {
Map<String, String> content = new HashMap<String, String>();
Map<String, String> content = new HashMap<>();
content.put("destination", destination);
content.put("binlogFile", position.getJournalName());
content.put("binlogOffest", String.valueOf(position.getPosition()));
@@ -326,7 +318,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
}
if (compareAll) {
Map<String, String> content = new HashMap<String, String>();
Map<String, String> content = new HashMap<>();
content.put("destination", destination);
content.put("binlogFile", position.getJournalName());
content.put("binlogOffest", String.valueOf(position.getPosition()));

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
View File

@@ -49,7 +49,7 @@ import com.alibaba.otter.canal.protocol.position.EntryPosition;
public class MemoryTableMeta implements TableMetaTSDB {
private Logger logger = LoggerFactory.getLogger(MemoryTableMeta.class);
private Map<List<String>, TableMeta> tableMetas = new ConcurrentHashMap<List<String>, TableMeta>();
private Map<List<String>, TableMeta> tableMetas = new ConcurrentHashMap<>();
private SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL);
public MemoryTableMeta(){
@@ -143,7 +143,7 @@ public class MemoryTableMeta implements TableMetaTSDB {
}
public Map<String, String> snapshot() {
Map<String, String> schemaDdls = new HashMap<String, String>();
Map<String, String> schemaDdls = new HashMap<>();
for (Schema schema : repository.getSchemas()) {
StringBuffer data = new StringBuffer(4 * 1024);
for (String table : schema.showTables()) {

32
parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java
View File

@@ -68,15 +68,10 @@ public class FileMixedLogPositionManager extends AbstractLogPositionManager {
this.period = period;
this.memoryLogPositionManager = memoryLogPositionManager;
this.dataFileCaches = MigrateMap.makeComputingMap(new Function<String, File>() {
public File apply(String destination) {
return getDataFile(destination);
}
});
this.dataFileCaches = MigrateMap.makeComputingMap(this::getDataFile);
this.executorService = Executors.newScheduledThreadPool(1);
this.persistTasks = Collections.synchronizedSet(new HashSet<String>());
this.persistTasks = Collections.synchronizedSet(new HashSet<>());
}
@Override
@@ -100,19 +95,16 @@ public class FileMixedLogPositionManager extends AbstractLogPositionManager {
}
// 启动定时工作任务
executorService.scheduleAtFixedRate(new Runnable() {
public void run() {
List<String> tasks = new ArrayList<String>(persistTasks);
for (String destination : tasks) {
try {
// 定时将内存中的最新值刷到file中，多次变更只刷一次
flushDataToFile(destination);
persistTasks.remove(destination);
} catch (Throwable e) {
// ignore
logger.error("period update" + destination + " curosr failed!", e);
}
executorService.scheduleAtFixedRate(() -> {
List<String> tasks = new ArrayList<>(persistTasks);
for (String destination : tasks) {
try {
// 定时将内存中的最新值刷到file中，多次变更只刷一次
flushDataToFile(destination);
persistTasks.remove(destination);
} catch (Throwable e) {
// ignore
logger.error("period update" + destination + " curosr failed!", e);
}
}
}, period, period, TimeUnit.MILLISECONDS);

13
parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java
View File

@@ -73,14 +73,11 @@ public class MixedLogPositionManager extends AbstractLogPositionManager {
@Override
public void persistLogPosition(final String destination, final LogPosition logPosition) throws CanalParseException {
memoryLogPositionManager.persistLogPosition(destination, logPosition);
executor.submit(new Runnable() {
public void run() {
try {
zooKeeperLogPositionManager.persistLogPosition(destination, logPosition);
} catch (Exception e) {
logger.error("ERROR # persist to zookeeper has an error", e);
}
executor.submit(() -> {
try {
zooKeeperLogPositionManager.persistLogPosition(destination, logPosition);
} catch (Exception e) {
logger.error("ERROR # persist to zookeeper has an error", e);
}
});
}

25
parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java
View File

@@ -50,7 +50,7 @@ public class PeriodMixedLogPositionManager extends AbstractLogPositionManager {
this.memoryLogPositionManager = memoryLogPositionManager;
this.zooKeeperLogPositionManager = zooKeeperLogPositionManager;
this.period = period;
this.persistTasks = Collections.synchronizedSet(new HashSet<String>());
this.persistTasks = Collections.synchronizedSet(new HashSet<>());
this.executorService = Executors.newScheduledThreadPool(1);
}
@@ -82,19 +82,16 @@ public class PeriodMixedLogPositionManager extends AbstractLogPositionManager {
}
// 启动定时工作任务
executorService.scheduleAtFixedRate(new Runnable() {
public void run() {
List<String> tasks = new ArrayList<String>(persistTasks);
for (String destination : tasks) {
try {
// 定时将内存中的最新值刷到zookeeper中，多次变更只刷一次
zooKeeperLogPositionManager.persistLogPosition(destination, getLatestIndexBy(destination));
persistTasks.remove(destination);
} catch (Throwable e) {
// ignore
logger.error("period update" + destination + " curosr failed!", e);
}
executorService.scheduleAtFixedRate(() -> {
List<String> tasks = new ArrayList<String>(persistTasks);
for (String destination : tasks) {
try {
// 定时将内存中的最新值刷到zookeeper中，多次变更只刷一次
zooKeeperLogPositionManager.persistLogPosition(destination, getLatestIndexBy(destination));
persistTasks.remove(destination);
} catch (Throwable e) {
// ignore
logger.error("period update" + destination + " curosr failed!", e);
}
}
}, period, period, TimeUnit.MILLISECONDS);

2
parse/src/main/java/com/alibaba/otter/canal/parse/support/HaAuthenticationInfo.java
View File

@@ -11,7 +11,7 @@ import java.util.List;
public class HaAuthenticationInfo {
private AuthenticationInfo master;
private List<AuthenticationInfo> slavers = new ArrayList<AuthenticationInfo>();
private List<AuthenticationInfo> slavers = new ArrayList<>();
public AuthenticationInfo getMaster() {
return master;

20
parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java
View File

@@ -41,18 +41,14 @@ public class MysqlBinlogParsePerformanceTest {
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();
}
final BlockingQueue<LogBuffer> buffer = new ArrayBlockingQueue<>(1024);
Thread thread = new Thread(() -> {
try {
consumer(buffer);
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
thread.start();

82
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/EventTransactionBufferTest.java
View File

@@ -25,29 +25,26 @@ public class EventTransactionBufferTest {
final int transactionSize = 5;
EventTransactionBuffer buffer = new EventTransactionBuffer();
buffer.setBufferSize(bufferSize);
buffer.setFlushCallback(new TransactionFlushCallback() {
buffer.setFlushCallback(transaction -> {
Assert.assertEquals(transactionSize, transaction.size());
System.out.println("\n>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
for (Entry data : transaction) {
public void flush(List<Entry> transaction) throws InterruptedException {
Assert.assertEquals(transactionSize, transaction.size());
System.out.println("\n>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
for (Entry data : transaction) {
CanalEntry.Header header = data.getHeader();
Date date = new Date(header.getExecuteTime());
SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
if (data.getEntryType() == EntryType.TRANSACTIONBEGIN
|| data.getEntryType() == EntryType.TRANSACTIONEND) {
System.out.println(data.getEntryType());
} else {
System.out.println(MessageFormat.format(messgae, new Object[] {
Thread.currentThread().getName(), header.getLogfileName(), header.getLogfileOffset(),
format.format(date), header.getSchemaName(), header.getTableName() }));
}
Header header = data.getHeader();
Date date = new Date(header.getExecuteTime());
SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
if (data.getEntryType() == EntryType.TRANSACTIONBEGIN
|| data.getEntryType() == EntryType.TRANSACTIONEND) {
System.out.println(data.getEntryType());
} else {
System.out.println(MessageFormat.format(messgae, new Object[] {
Thread.currentThread().getName(), header.getLogfileName(), header.getLogfileOffset(),
format.format(date), header.getSchemaName(), header.getTableName() }));
}
System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n");
}
System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n");
});
buffer.start();
@@ -73,35 +70,32 @@ public class EventTransactionBufferTest {
final int bufferSize = 64;
EventTransactionBuffer buffer = new EventTransactionBuffer();
buffer.setBufferSize(bufferSize);
buffer.setFlushCallback(new TransactionFlushCallback() {
buffer.setFlushCallback(transaction -> {
Assert.assertEquals(bufferSize, transaction.size());
System.out.println("\n>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
for (Entry data : transaction) {
public void flush(List<Entry> transaction) throws InterruptedException {
Assert.assertEquals(bufferSize, transaction.size());
System.out.println("\n>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
for (Entry data : transaction) {
CanalEntry.Header header = data.getHeader();
Date date = new Date(header.getExecuteTime());
SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
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.getEntryType());
} else {
System.out.println(MessageFormat.format(messgae, new Object[] {
Thread.currentThread().getName(), header.getLogfileName(), header.getLogfileOffset(),
format.format(date), header.getSchemaName(), header.getTableName() }));
}
Header header = data.getHeader();
Date date = new Date(header.getExecuteTime());
SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
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.getEntryType());
} else {
System.out.println(MessageFormat.format(messgae, new Object[] {
Thread.currentThread().getName(), header.getLogfileName(), header.getLogfileOffset(),
format.format(date), header.getSchemaName(), header.getTableName() }));
}
System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n");
}
System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n");
});
buffer.start();

2
parse/src/test/java/com/alibaba/otter/canal/parse/index/MetaLogPositionManagerTest.java
View File

@@ -82,6 +82,6 @@ public class MetaLogPositionManagerTest extends AbstractLogPositionManagerTest {
LogPosition end = new LogPosition();
end.setIdentity(new LogIdentity(new InetSocketAddress(MYSQL_ADDRESS, 3306), 1234L));
end.setPostion(new EntryPosition("mysql-bin.000000" + (number + 1), 106L, (new Date().getTime()) + 1000 * 1000L));
return new PositionRange<LogPosition>(start, end);
return new PositionRange<>(start, end);
}
}