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

Refactor metrics code.

Browse Source
This commit is contained in:
Chuanyi Li L
2018-08-07 17:17:33 +08:00
parent b5a1898cec
commit 40c3e65341
17 changed files with 462 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -8,7 +8,6 @@ import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.common.utils.SerializedLongAdder;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.lang.math.RandomUtils;

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

@@ -1,8 +1,8 @@
package com.alibaba.otter.canal.parse.inbound.mysql;
import java.nio.charset.Charset;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.common.utils.SerializedLongAdder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,7 +35,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
protected boolean filterRows = false;
protected boolean filterTableError = false;
protected boolean useDruidDdlFilter = true;
private final SerializedLongAdder eventsPublishBlockingTime = new SerializedLongAdder(0L);
private final AtomicLong eventsPublishBlockingTime = new AtomicLong(0L);
protected BinlogParser buildParser() {
LogEventConvert convert = new LogEventConvert();
@@ -192,7 +192,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
}
}
public SerializedLongAdder getEventsPublishBlockingTime() {
public AtomicLong getEventsPublishBlockingTime() {
return this.eventsPublishBlockingTime;
}

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

@@ -7,8 +7,8 @@ import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.common.utils.SerializedLongAdder;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,7 +50,7 @@ public class MysqlConnection implements ErosaConnection {
private AuthenticationInfo authInfo;
protected int connTimeout = 5 * 1000; // 5秒
protected int soTimeout = 60 * 60 * 1000; // 1小时
private SerializedLongAdder receivedBinlogBytes;
private AtomicLong receivedBinlogBytes;
public MysqlConnection(){
}
@@ -461,7 +461,7 @@ public class MysqlConnection implements ErosaConnection {
private void accumulateReceivedBytes(long x) {
if (receivedBinlogBytes != null) {
receivedBinlogBytes.add(x);
receivedBinlogBytes.addAndGet(x);
}
}
@@ -606,7 +606,7 @@ public class MysqlConnection implements ErosaConnection {
this.authInfo = authInfo;
}
public void setReceivedBinlogBytes(SerializedLongAdder receivedBinlogBytes) {
public void setReceivedBinlogBytes(AtomicLong receivedBinlogBytes) {
this.receivedBinlogBytes = receivedBinlogBytes;
}

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

@@ -10,7 +10,6 @@ import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.common.utils.SerializedLongAdder;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.CollectionUtils;
@@ -74,7 +73,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值
// instance received binlog bytes
private final SerializedLongAdder receivedBinlogBytes = new SerializedLongAdder(0L);
private final AtomicLong receivedBinlogBytes = new AtomicLong(0L);
protected ErosaConnection buildErosaConnection() {
return buildMysqlConnection(this.runningInfo);
@@ -517,7 +516,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
private Long findTransactionBeginPosition(ErosaConnection mysqlConnection, final EntryPosition entryPosition)
throws IOException {
// 针对开始的第一条为非Begin记录，需要从该binlog扫描
final AtomicLong preTransactionStartPosition = new AtomicLong(0L);
final java.util.concurrent.atomic.AtomicLong preTransactionStartPosition = new java.util.concurrent.atomic.AtomicLong(0L);
mysqlConnection.reconnect();
mysqlConnection.seek(entryPosition.getJournalName(), 4L, new SinkFunction<LogEvent>() {
@@ -918,7 +917,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
public SerializedLongAdder getReceivedBinlogBytes() {
public AtomicLong getReceivedBinlogBytes() {
return this.receivedBinlogBytes;
}

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

@@ -2,9 +2,9 @@ package com.alibaba.otter.canal.parse.inbound.mysql;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;
import com.alibaba.otter.canal.common.utils.SerializedLongAdder;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
@@ -65,7 +65,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
private ExecutorService stageExecutor;
private String destination;
private volatile CanalParseException exception;
private SerializedLongAdder eventsPublishBlockingTime;
private AtomicLong eventsPublishBlockingTime;
public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert,
EventTransactionBuffer transactionBuffer, String destination){
@@ -176,7 +176,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
}
disruptorMsgBuffer.publish(next);
if (fullTimes > 0) {
eventsPublishBlockingTime.add(System.nanoTime() - blockingStart);
eventsPublishBlockingTime.addAndGet(System.nanoTime() - blockingStart);
}
break;
} catch (InsufficientCapacityException e) {
@@ -451,7 +451,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
this.connection = connection;
}
public void setEventsPublishBlockingTime(SerializedLongAdder eventsPublishBlockingTime) {
public void setEventsPublishBlockingTime(AtomicLong eventsPublishBlockingTime) {
this.eventsPublishBlockingTime = eventsPublishBlockingTime;
}