聊聊BinaryLogClient的connect
序
本文主要研究一下BinaryLogClient的connect
connect(timeout)
mysql-binlog-connector-java-0.20.1/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java
public class BinaryLogClient implements BinaryLogClientMXBean {
//......
public void connect(final long timeout) throws IOException, TimeoutException {
final CountDownLatch countDownLatch = new CountDownLatch(1);
AbstractLifecycleListener connectListener = new AbstractLifecycleListener() {
@Override
public void onConnect(BinaryLogClient client) {
countDownLatch.countDown();
}
};
registerLifecycleListener(connectListener);
final AtomicReference<IOException> exceptionReference = new AtomicReference<IOException>();
Runnable runnable = new Runnable() {
@Override
public void run() {
try {
setConnectTimeout(timeout);
connect();
} catch (IOException e) {
exceptionReference.set(e);
countDownLatch.countDown(); // making sure we don't end up waiting whole "timeout"
}
}
};
newNamedThread(runnable, "blc-" + hostname + ":" + port).start();
boolean started = false;
try {
started = countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
if (logger.isLoggable(Level.WARNING)) {
logger.log(Level.WARNING, e.getMessage());
}
}
unregisterLifecycleListener(connectListener);
if (exceptionReference.get() != null) {
throw exceptionReference.get();
}
if (!started) {
try {
terminateConnect();
} finally {
throw new TimeoutException("BinaryLogClient was unable to connect in " + timeout + "ms");
}
}
}
//......
}
- 带timeout参数的connect方法使用setConnectTimeout方法设置了超时时间，然后再执行无参的connect方法；之后在捕获IOException的时候更新exceptionReference以及触发countDownLatch.countDown()，之后抛出exceptionReference.get()
connect
mysql-binlog-connector-java-0.20.1/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java
public class BinaryLogClient implements BinaryLogClientMXBean {
//......
public void connect() throws IOException {
if (!connectLock.tryLock()) {
throw new IllegalStateException("BinaryLogClient is already connected");
}
boolean notifyWhenDisconnected = false;
try {
Callable cancelDisconnect = null;
try {
try {
long start = System.currentTimeMillis();
channel = openChannel();
if (connectTimeout > 0 && !isKeepAliveThreadRunning()) {
cancelDisconnect = scheduleDisconnectIn(connectTimeout -
(System.currentTimeMillis() - start));
}
if (channel.getInputStream().peek() == -1) {
throw new EOFException();
}
} catch (IOException e) {
throw new IOException("Failed to connect to MySQL on " + hostname + ":" + port +
". Please make sure it's running.", e);
}
GreetingPacket greetingPacket = receiveGreeting();
authenticate(greetingPacket);
connectionId = greetingPacket.getThreadId();
if ("".equals(binlogFilename)) {
synchronized (gtidSetAccessLock) {
if (gtidSet != null && "".equals(gtidSet.toString()) && gtidSetFallbackToPurged) {
gtidSet = new GtidSet(fetchGtidPurged());
}
}
}
if (binlogFilename == null) {
fetchBinlogFilenameAndPosition();
}
if (binlogPosition < 4) {
if (logger.isLoggable(Level.WARNING)) {
logger.warning("Binary log position adjusted from " + binlogPosition + " to " + 4);
}
binlogPosition = 4;
}
ChecksumType checksumType = fetchBinlogChecksum();
if (checksumType != ChecksumType.NONE) {
confirmSupportOfChecksum(checksumType);
}
if (heartbeatInterval > 0) {
enableHeartbeat();
}
gtid = null;
tx = false;
requestBinaryLogStream();
} catch (IOException e) {
disconnectChannel();
throw e;
} finally {
if (cancelDisconnect != null) {
try {
cancelDisconnect.call();
} catch (Exception e) {
if (logger.isLoggable(Level.WARNING)) {
logger.warning("/"" + e.getMessage() +
"/" was thrown while canceling scheduled disconnect call");
}
}
}
}
connected = true;
notifyWhenDisconnected = true;
if (logger.isLoggable(Level.INFO)) {
String position;
synchronized (gtidSetAccessLock) {
position = gtidSet != null ? gtidSet.toString() : binlogFilename + "/" + binlogPosition;
}
logger.info("Connected to " + hostname + ":" + port + " at " + position +
" (" + (blocking ? "sid:" + serverId + ", " : "") + "cid:" + connectionId + ")");
}
for (LifecycleListener lifecycleListener : lifecycleListeners) {
lifecycleListener.onConnect(this);
}
if (keepAlive && !isKeepAliveThreadRunning()) {
spawnKeepAliveThread();
}
ensureEventDataDeserializer(EventType.ROTATE, RotateEventDataDeserializer.class);
synchronized (gtidSetAccessLock) {
if (gtidSet != null) {
ensureEventDataDeserializer(EventType.GTID, GtidEventDataDeserializer.class);
ensureEventDataDeserializer(EventType.QUERY, QueryEventDataDeserializer.class);
}
}
listenForEventPackets();
} finally {
connectLock.unlock();
if (notifyWhenDisconnected) {
for (LifecycleListener lifecycleListener : lifecycleListeners) {
lifecycleListener.onDisconnect(this);
}
}
}
}
//......
}
- connect方法先执行openChannel方法，之后在connectTimeout大于0且keepAliveThreadRunning为false时通过scheduleDisconnectIn调度执行超时逻辑；之后执行receiveGreeting获取greetingPacket，然后执行authenticate(greetingPacket)，之后获取connectionId及设置gtidSet；在binlogFilename为null时执行fetchBinlogFilenameAndPosition；之后执行fetchBinlogChecksum，对于heartbeatInterval大于0的执行enableHeartbeat，最后执行requestBinaryLogStream；若出现IOException则执行disconnectChannel并抛出异常；连接成功之后回调lifecycleListeners的onConnect，对于需要keepAlive的执行spawnKeepAliveThread；最后执行listenForEventPackets
scheduleDisconnectIn
mysql-binlog-connector-java-0.20.1/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java
public class BinaryLogClient implements BinaryLogClientMXBean {
//......
private Callable scheduleDisconnectIn(final long timeout) {
final BinaryLogClient self = this;
final CountDownLatch connectLatch = new CountDownLatch(1);
final Thread thread = newNamedThread(new Runnable() {
@Override
public void run() {
try {
connectLatch.await(timeout, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
if (logger.isLoggable(Level.WARNING)) {
logger.log(Level.WARNING, e.getMessage());
}
}
if (connectLatch.getCount() != 0) {
if (logger.isLoggable(Level.WARNING)) {
logger.warning("Failed to establish connection in " + timeout + "ms. " +
"Forcing disconnect.");
}
try {
self.disconnectChannel();
} catch (IOException e) {
if (logger.isLoggable(Level.WARNING)) {
logger.log(Level.WARNING, e.getMessage());
}
}
}
}
}, "blc-disconnect-" + hostname + ":" + port);
thread.start();
return new Callable() {
public Object call() throws Exception {
connectLatch.countDown();
thread.join();
return null;
}
};
}
private void disconnectChannel() throws IOException {
connected = false;
if (channel != null && channel.isOpen()) {
channel.close();
}
}
//......
}
- scheduleDisconnectIn方法主要是通过CountDownLatch来执行超时等待，超时的话执行disconnectChannel方法，它会关闭channel
receiveGreeting
mysql-binlog-connector-java-0.20.1/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java
public class BinaryLogClient implements BinaryLogClientMXBean {
//......
private GreetingPacket receiveGreeting() throws IOException {
byte[] initialHandshakePacket = channel.read();
if (initialHandshakePacket[0] == (byte) 0xFF /* error */) {
byte[] bytes = Arrays.copyOfRange(initialHandshakePacket, 1, initialHandshakePacket.length);
ErrorPacket errorPacket = new ErrorPacket(bytes);
throw new ServerException(errorPacket.getErrorMessage(), errorPacket.getErrorCode(),
errorPacket.getSqlState());
}
return new GreetingPacket(initialHandshakePacket);
}
private void fetchBinlogFilenameAndPosition() throws IOException {
ResultSetRowPacket[] resultSet;
channel.write(new QueryCommand("show master status"));
resultSet = readResultSet();
if (resultSet.length == 0) {
throw new IOException("Failed to determine binlog filename/position");
}
ResultSetRowPacket resultSetRow = resultSet[0];
binlogFilename = resultSetRow.getValue(0);
binlogPosition = Long.parseLong(resultSetRow.getValue(1));
}
private ChecksumType fetchBinlogChecksum() throws IOException {
channel.write(new QueryCommand("show global variables like 'binlog_checksum'"));
ResultSetRowPacket[] resultSet = readResultSet();
if (resultSet.length == 0) {
return ChecksumType.NONE;
}
return ChecksumType.valueOf(resultSet[0].getValue(1).toUpperCase());
}
private void enableHeartbeat() throws IOException {
channel.write(new QueryCommand("set @master_heartbeat_period=" + heartbeatInterval * 1000000));
byte[] statementResult = channel.read();
if (statementResult[0] == (byte) 0xFF /* error */) {
byte[] bytes = Arrays.copyOfRange(statementResult, 1, statementResult.length);
ErrorPacket errorPacket = new ErrorPacket(bytes);
throw new ServerException(errorPacket.getErrorMessage(), errorPacket.getErrorCode(),
errorPacket.getSqlState());
}
}
private void requestBinaryLogStream() throws IOException {
long serverId = blocking ? this.serverId : 0; // http://bugs.mysql.com/bug.php?id=71178
Command dumpBinaryLogCommand;
synchronized (gtidSetAccessLock) {
if (gtidSet != null) {
dumpBinaryLogCommand = new DumpBinaryLogGtidCommand(serverId,
useBinlogFilenamePositionInGtidMode ? binlogFilename : "",
useBinlogFilenamePositionInGtidMode ? binlogPosition : 4,
gtidSet);
} else {
dumpBinaryLogCommand = new DumpBinaryLogCommand(serverId, binlogFilename, binlogPosition);
}
}
channel.write(dumpBinaryLogCommand);
}
private void spawnKeepAliveThread() {
final ExecutorService threadExecutor =
Executors.newSingleThreadExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable runnable) {
return newNamedThread(runnable, "blc-keepalive-" + hostname + ":" + port);
}
});
try {
keepAliveThreadExecutorLock.lock();
threadExecutor.submit(new Runnable() {
@Override
public void run() {
while (!threadExecutor.isShutdown()) {
try {
Thread.sleep(keepAliveInterval);
} catch (InterruptedException e) {
// expected in case of disconnect
}
if (threadExecutor.isShutdown()) {
return;
}
boolean connectionLost = false;
if (heartbeatInterval > 0) {
connectionLost = System.currentTimeMillis() - eventLastSeen > keepAliveInterval;
} else {
try {
channel.write(new PingCommand());
} catch (IOException e) {
connectionLost = true;
}
}
if (connectionLost) {
if (logger.isLoggable(Level.INFO)) {
logger.info("Trying to restore lost connection to " + hostname + ":" + port);
}
try {
terminateConnect();
connect(connectTimeout);
} catch (Exception ce) {
if (logger.isLoggable(Level.WARNING)) {
logger.warning("Failed to restore connection to " + hostname + ":" + port +
". Next attempt in " + keepAliveInterval + "ms");
}
}
}
}
}
});
keepAliveThreadExecutor = threadExecutor;
} finally {
keepAliveThreadExecutorLock.unlock();
}
}
//......
}
-
receiveGreeting方法接收initialHandshakePacket；fetchBinlogFilenameAndPosition方法发送
show master status命令，然后解析binlogFilename及binlogPosition；fetchBinlogChecksum方法发送show global variables like 'binlog_checksum'命令；enableHeartbeat方法发送set @master_heartbeat_period命令；requestBinaryLogStream方法发送DumpBinaryLogGtidCommand命令；spawnKeepAliveThread方法会调度执行keepAliveThread，它主体是while循环，然后通过Thread.sleep(keepAliveInterval)来实现间隔，然后定时发送PingCommand命令，对于connectionLost的执行terminateConnect及connect方法
listenForEventPackets
mysql-binlog-connector-java-0.20.1/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java
public class BinaryLogClient implements BinaryLogClientMXBean {
//......
private void listenForEventPackets() throws IOException {
ByteArrayInputStream inputStream = channel.getInputStream();
boolean completeShutdown = false;
try {
while (inputStream.peek() != -1) {
int packetLength = inputStream.readInteger(3);
inputStream.skip(1); // 1 byte for sequence
int marker = inputStream.read();
if (marker == 0xFF) {
ErrorPacket errorPacket = new ErrorPacket(inputStream.read(packetLength - 1));
throw new ServerException(errorPacket.getErrorMessage(), errorPacket.getErrorCode(),
errorPacket.getSqlState());
}
if (marker == 0xFE && !blocking) {
completeShutdown = true;
break;
}
Event event;
try {
event = eventDeserializer.nextEvent(packetLength == MAX_PACKET_LENGTH ?
new ByteArrayInputStream(readPacketSplitInChunks(inputStream, packetLength - 1)) :
inputStream);
if (event == null) {
throw new EOFException();
}
} catch (Exception e) {
Throwable cause = e instanceof EventDataDeserializationException ? e.getCause() : e;
if (cause instanceof EOFException || cause instanceof SocketException) {
throw e;
}
if (isConnected()) {
for (LifecycleListener lifecycleListener : lifecycleListeners) {
lifecycleListener.onEventDeserializationFailure(this, e);
}
}
continue;
}
if (isConnected()) {
eventLastSeen = System.currentTimeMillis();
updateGtidSet(event);
notifyEventListeners(event);
updateClientBinlogFilenameAndPosition(event);
}
}
} catch (Exception e) {
if (isConnected()) {
for (LifecycleListener lifecycleListener : lifecycleListeners) {
lifecycleListener.onCommunicationFailure(this, e);
}
}
} finally {
if (isConnected()) {
if (completeShutdown) {
disconnect(); // initiate complete shutdown sequence (which includes keep alive thread)
} else {
disconnectChannel();
}
}
}
}
//......
}
- listenForEventPackets方法读取channel.getInputStream()，并通过eventDeserializer.nextEvent解析为event，然后通过notifyEventListeners方法将event通知eventListener
小结
connect方法先执行openChannel方法，之后在connectTimeout大于0且keepAliveThreadRunning为false时通过scheduleDisconnectIn调度执行超时逻辑；之后执行receiveGreeting获取greetingPacket，然后执行authenticate(greetingPacket)，之后获取connectionId及设置gtidSet；在binlogFilename为null时执行fetchBinlogFilenameAndPosition；之后执行fetchBinlogChecksum，对于heartbeatInterval大于0的执行enableHeartbeat，最后执行requestBinaryLogStream；若出现IOException则执行disconnectChannel并抛出异常；连接成功之后回调lifecycleListeners的onConnect，对于需要keepAlive的执行spawnKeepAliveThread；最后执行listenForEventPackets
doc
- BinaryLogClient
正文到此结束
- 本文标签： update git 时间 参数 message id ssh ip GitHub REST Listeners Master stream java heartbeat IDE list mysql-bin IO ResultSet rmi mina sql value http executor Statement mysql tab cat parse src ORM CountDownLatch authenticate tar 解析 client bean synchronized ssl CTO Connection Service equals PHP https CEO final Atom bug
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
近期评论
-
666
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
