From 48c108806ae4aba3d04c11cf7d9e257ca9609803 Mon Sep 17 00:00:00 2001 From: luoyaogui Date: Tue, 27 Jun 2017 15:02:45 +0800 Subject: [PATCH 1/9] Update SocketChannel.java --- .../driver/mysql/socket/SocketChannel.java | 40 ++++++++----------- 1 file changed, 17 insertions(+), 23 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java index e4d1bf65..1fbb7c7b 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java @@ -7,7 +7,6 @@ import io.netty.channel.Channel; import java.io.IOException; import java.net.SocketAddress; -import java.nio.ByteBuffer; /** * @author luoyaogui @@ -17,18 +16,13 @@ import java.nio.ByteBuffer; public class SocketChannel { private Channel channel = null; private Object lock = new Object(); - private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(1024*1024*5);//缓存大小 + private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(1024*1024);//缓存大小 public Channel getChannel() { return channel; } - public void setChannel(Channel channel,boolean notify) { + public void setChannel(Channel channel) { this.channel = channel; - if(notify){//是否需要通知，主要时channel不可用时 - synchronized(this){ - notifyAll(); - } - } } public void writeCache(ByteBuf buf){ @@ -36,9 +30,6 @@ public class SocketChannel { cache.discardReadBytes();//回收内存 cache.writeBytes(buf); } - synchronized(this){ - notifyAll(); - } } public void writeChannel(byte[]... buf) throws IOException { if(channel != null && channel.isWritable()) @@ -46,19 +37,22 @@ public class SocketChannel { else throw new IOException("write failed ! please checking !"); } - public int read(ByteBuffer buffer) throws IOException { - if(null == channel) - throw new IOException("socket has Interrupted !"); - if(cache.readableBytes() < buffer.remaining()){ - synchronized(this){ - try {wait();} catch (InterruptedException e) {} + public byte[] read(int readSize) throws IOException { + do{ + if(readSize > cache.readableBytes()){ + if(null == channel) + throw new IOException("socket has Interrupted !"); + synchronized (this) { + try { wait(100); } catch (InterruptedException e) {} + } + } else { + byte[] back = new byte[readSize]; + synchronized (lock) { + cache.readBytes(back); + } + return back; } - }else{ - synchronized (lock){ - cache.readBytes(buffer); - } - } - return 0; + }while(true); } public boolean isConnected() { return channel!=null?true:false; From 9ea88c78637b1063d8c0e4ca4cea44e6a1b62d31 Mon Sep 17 00:00:00 2001 From: luoyaogui Date: Tue, 27 Jun 2017 15:03:16 +0800 Subject: [PATCH 2/9] Update SocketChannelPool.java --- .../canal/parse/driver/mysql/socket/SocketChannelPool.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java index 3d6d17c7..e7d791a3 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java @@ -54,7 +54,7 @@ public abstract class SocketChannelPool { @Override public void operationComplete(ChannelFuture arg0) throws Exception { if(arg0.isSuccess()) - socket.setChannel(arg0.channel(),false); + socket.setChannel(arg0.channel()); synchronized (socket) { socket.notify(); } @@ -73,7 +73,7 @@ public abstract class SocketChannelPool { private SocketChannel socket=null; @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { - socket.setChannel(null,true); + socket.setChannel(null); chManager.remove(ctx.channel());//移除 } @Override From b6090da0e5123b843dd9cf50e62cdc3d33f22493 Mon Sep 17 00:00:00 2001 From: luoyaogui Date: Tue, 27 Jun 2017 15:03:53 +0800 Subject: [PATCH 3/9] Update PacketManager.java --- .../driver/mysql/utils/PacketManager.java | 39 ++++--------------- 1 file changed, 8 insertions(+), 31 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java index fa3ca4c1..797918e4 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java @@ -1,7 +1,6 @@ package com.alibaba.otter.canal.parse.driver.mysql.utils; import java.io.IOException; -import java.nio.ByteBuffer; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; @@ -10,48 +9,26 @@ public abstract class PacketManager { public static HeaderPacket readHeader(SocketChannel ch, int len) throws IOException { HeaderPacket header = new HeaderPacket(); - header.fromBytes(readBytesAsBuffer(ch, len).array()); + header.fromBytes(ch.read(len)); return header; } - public static ByteBuffer readBytesAsBuffer(SocketChannel ch, int len) throws IOException { - ByteBuffer buffer = ByteBuffer.allocate(len); - while (buffer.hasRemaining()) { - int readNum = ch.read(buffer); - if (readNum == -1) { - throw new IOException("Unexpected End Stream"); - } - } - return buffer; - } - public static byte[] readBytes(SocketChannel ch, int len) throws IOException { - return readBytesAsBuffer(ch, len).array(); + return ch.read(len); } - /** - * Since We r using blocking IO, so we will just write once and assert the - * length to simplify the read operation.
- * If the block write doesn't work as we expected, we will change this - * implementation as per the result. - * - * @param ch - * @param len - * @return - * @throws IOException - */ - public static void write(SocketChannel ch, byte[]... srcs) throws IOException { + public static void writePkg(SocketChannel ch, byte[]... srcs) throws IOException { ch.writeChannel(srcs); } - - public static void write(SocketChannel ch, byte[] body) throws IOException { - write(ch, body, (byte) 0); + + public static void writeBody(SocketChannel ch, byte[] body) throws IOException { + writeBody0(ch, body, (byte) 0); } - public static void write(SocketChannel ch, byte[] body, byte packetSeqNumber) throws IOException { + public static void writeBody0(SocketChannel ch, byte[] body, byte packetSeqNumber) throws IOException { HeaderPacket header = new HeaderPacket(); header.setPacketBodyLength(body.length); header.setPacketSequenceNumber(packetSeqNumber); - write(ch, header.toBytes(),body); + ch.writeChannel(header.toBytes(),body); } } From 525a00ea17bc8cd929aff154e7ea1525f9c2d9b9 Mon Sep 17 00:00:00 2001 From: luoyaogui Date: Tue, 27 Jun 2017 15:04:39 +0800 Subject: [PATCH 4/9] Update MysqlConnector.java --- .../canal/parse/driver/mysql/MysqlConnector.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java index 89fe1718..e22d3be4 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java @@ -2,6 +2,9 @@ package com.alibaba.otter.canal.parse.driver.mysql; import java.io.IOException; import java.net.InetSocketAddress; +import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; +import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannelPool; + import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.lang.exception.ExceptionUtils; @@ -14,8 +17,6 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QuitCommandPack import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.HandshakeInitializationPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.Reply323Packet; -import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; -import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannelPool; import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter; import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager; @@ -101,8 +102,7 @@ public class MysqlConnector { MysqlUpdateExecutor executor = new MysqlUpdateExecutor(connector); executor.update("KILL CONNECTION " + connectionId); } catch (Exception e) { - // 忽略具体异常 - logger.info("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e)); + throw new IOException("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e)); } finally { if (connector != null) { connector.disconnect(); @@ -140,7 +140,7 @@ public class MysqlConnector { HeaderPacket quitHeader = new HeaderPacket(); quitHeader.setPacketBodyLength(cmdBody.length); quitHeader.setPacketSequenceNumber((byte) 0x00); - PacketManager.write(channel,quitHeader.toBytes(),cmdBody); + PacketManager.writePkg(channel, quitHeader.toBytes(),cmdBody); } private void negotiate(SocketChannel channel) throws IOException { @@ -177,7 +177,7 @@ public class MysqlConnector { h.setPacketBodyLength(clientAuthPkgBody.length); h.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1)); - PacketManager.write(channel,h.toBytes(), clientAuthPkgBody); + PacketManager.writePkg(channel, h.toBytes(), clientAuthPkgBody); logger.info("client authentication packet is sent out."); // check auth result @@ -213,7 +213,7 @@ public class MysqlConnector { h323.setPacketBodyLength(b323Body.length); h323.setPacketSequenceNumber((byte) (packetSequenceNumber + 1)); - PacketManager.write(channel, h323.toBytes(), b323Body); + PacketManager.writePkg(channel, h323.toBytes(), b323Body); logger.info("client 323 authentication packet is sent out."); // check auth result HeaderPacket header = PacketManager.readHeader(channel, 4); From 523130cf8b866abe3f497e3258e2e9f180fd7f1f Mon Sep 17 00:00:00 2001 From: luoyaogui Date: Tue, 27 Jun 2017 15:05:00 +0800 Subject: [PATCH 5/9] Update MysqlQueryExecutor.java --- .../canal/parse/driver/mysql/MysqlQueryExecutor.java | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java index ef7d3777..feb10d2c 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java @@ -1,6 +1,7 @@ package com.alibaba.otter.canal.parse.driver.mysql; import java.io.IOException; +import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; import java.util.ArrayList; import java.util.List; @@ -11,7 +12,6 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetHeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.RowDataPacket; -import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager; /** @@ -51,7 +51,7 @@ public class MysqlQueryExecutor { QueryCommandPacket cmd = new QueryCommandPacket(); cmd.setQueryString(queryString); byte[] bodyBytes = cmd.toBytes(); - PacketManager.write(channel, bodyBytes); + PacketManager.writeBody(channel, bodyBytes); byte[] body = readNextPacket(); if (body[0] < 0) { @@ -82,9 +82,7 @@ public class MysqlQueryExecutor { rowDataPacket.fromBytes(body); rowData.add(rowDataPacket); } - //未知，不知道是否需要锁定 - //channel.lock();//锁定读 - + ResultSetPacket resultSet = new ResultSetPacket(); resultSet.getFieldDescriptors().addAll(fields); for (RowDataPacket r : rowData) { From 3dba40062441a1a8e72f76fbf9e6baa66b4e7986 Mon Sep 17 00:00:00 2001 From: luoyaogui Date: Tue, 27 Jun 2017 15:05:33 +0800 Subject: [PATCH 6/9] Update MysqlUpdateExecutor.java --- .../parse/driver/mysql/MysqlUpdateExecutor.java | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlUpdateExecutor.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlUpdateExecutor.java index a604b71e..25c4dd3c 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlUpdateExecutor.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlUpdateExecutor.java @@ -1,6 +1,7 @@ package com.alibaba.otter.canal.parse.driver.mysql; import java.io.IOException; +import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -8,7 +9,6 @@ import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QueryCommandPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.OKPacket; -import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager; /** @@ -21,34 +21,33 @@ public class MysqlUpdateExecutor { private static final Logger logger = LoggerFactory.getLogger(MysqlUpdateExecutor.class); - private SocketChannel channel; + private MysqlConnector connector; public MysqlUpdateExecutor(MysqlConnector connector){ if (!connector.isConnected()) { throw new RuntimeException("should execute connector.connect() first"); } - this.channel = connector.getChannel(); + this.connector = connector; } - public MysqlUpdateExecutor(SocketChannel ch){ + /* public MysqlUpdateExecutor(SocketChannel ch){ this.channel = ch; - } + }*/ public OKPacket update(String updateString) throws IOException { QueryCommandPacket cmd = new QueryCommandPacket(); cmd.setQueryString(updateString); byte[] bodyBytes = cmd.toBytes(); - PacketManager.write(channel, bodyBytes); + PacketManager.writeBody(connector.getChannel(), bodyBytes); logger.debug("read update result..."); - byte[] body = PacketManager.readBytes(channel, PacketManager.readHeader(channel, 4).getPacketBodyLength()); + byte[] body = PacketManager.readBytes(connector.getChannel(), PacketManager.readHeader(connector.getChannel(), 4).getPacketBodyLength()); if (body[0] < 0) { ErrorPacket packet = new ErrorPacket(); packet.fromBytes(body); throw new IOException(packet + "\n with command: " + updateString); } - //channel.lock();//锁定读 OKPacket packet = new OKPacket(); packet.fromBytes(body); From 0bd4a1146ce0814934174888184c2d5050373ab4 Mon Sep 17 00:00:00 2001 From: luoyaogui Date: Tue, 27 Jun 2017 15:07:07 +0800 Subject: [PATCH 7/9] Update DirectLogFetcherTest.java --- .../com/alibaba/otter/canal/parse/DirectLogFetcherTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java index 0323a733..7b67ebea 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java @@ -94,6 +94,6 @@ public class DirectLogFetcherTest { HeaderPacket binlogDumpHeader = new HeaderPacket(); binlogDumpHeader.setPacketBodyLength(cmdBody.length); binlogDumpHeader.setPacketSequenceNumber((byte) 0x00); - PacketManager.write(connector.getChannel(), binlogDumpHeader.toBytes(),cmdBody); + PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody); } } From 0b2f37a2720947131049f938bc111f1040c63d7b Mon Sep 17 00:00:00 2001 From: luoyaogui Date: Tue, 27 Jun 2017 15:07:52 +0800 Subject: [PATCH 8/9] Update MysqlConnection.java --- .../parse/inbound/mysql/MysqlConnection.java | 44 ++----------------- 1 file changed, 4 insertions(+), 40 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index eb270c92..74e561b1 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -2,11 +2,9 @@ package com.alibaba.otter.canal.parse.inbound.mysql; import java.io.IOException; import java.net.InetSocketAddress; -import java.nio.ByteBuffer; import java.nio.charset.Charset; import java.util.List; -import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -25,8 +23,6 @@ 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; -import com.taobao.tddl.dbsync.binlog.LogPosition; -import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; public class MysqlConnection implements ErosaConnection { @@ -37,7 +33,6 @@ public class MysqlConnection implements ErosaConnection { private Charset charset = Charset.forName("UTF-8"); private BinlogFormat binlogFormat; private BinlogImage binlogImage; - private int binlogChecksum; public MysqlConnection(){ } @@ -83,7 +78,7 @@ public class MysqlConnection implements ErosaConnection { */ public void seek(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { updateSettings(); - loadBinlogChecksum(); + sendBinlogDump(binlogfilename, binlogPosition); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); fetcher.start(connector.getChannel()); @@ -93,8 +88,6 @@ public class MysqlConnection implements ErosaConnection { decoder.handle(LogEvent.QUERY_EVENT); decoder.handle(LogEvent.XID_EVENT); LogContext context = new LogContext(); - context.setLogPosition(new LogPosition(binlogfilename)); - context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); while (fetcher.fetch()) { LogEvent event = null; event = decoder.decode(fetcher, context); @@ -111,14 +104,11 @@ public class MysqlConnection implements ErosaConnection { public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { updateSettings(); - loadBinlogChecksum(); sendBinlogDump(binlogfilename, binlogPosition); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); fetcher.start(connector.getChannel()); LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); - context.setLogPosition(new LogPosition(binlogfilename)); - context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); while (fetcher.fetch()) { LogEvent event = null; event = decoder.decode(fetcher, context); @@ -148,8 +138,7 @@ public class MysqlConnection implements ErosaConnection { HeaderPacket binlogDumpHeader = new HeaderPacket(); binlogDumpHeader.setPacketBodyLength(cmdBody.length); binlogDumpHeader.setPacketSequenceNumber((byte) 0x00); - PacketManager.write(connector.getChannel(), binlogDumpHeader.toBytes(), - cmdBody); + PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(),cmdBody); connector.setDumping(true); } @@ -205,12 +194,9 @@ public class MysqlConnection implements ErosaConnection { // 如果不设置会出现错误： Slave can not handle replication events with the // checksum that master is configured to log // 但也不能乱设置，需要和mysql server的checksum配置一致，不然RotateLogEvent会出现乱码 - // '@@global.binlog_checksum'需要去掉单引号,在mysql 5.6.29下导致master退出 - update("set @master_binlog_checksum= @@global.binlog_checksum"); + update("set @master_binlog_checksum= '@@global.binlog_checksum'"); } catch (Exception e) { - if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) { - logger.warn(ExceptionUtils.getFullStackTrace(e)); - } + logger.warn(ExceptionUtils.getFullStackTrace(e)); } try { @@ -268,28 +254,6 @@ public class MysqlConnection implements ErosaConnection { } } - /** - * 获取主库checksum信息 - * https://dev.mysql.com/doc/refman/5.6/en/replication-options - * -binary-log.html#option_mysqld_binlog-checksum - */ - private void loadBinlogChecksum() { - ResultSetPacket rs = null; - try { - rs = query("select @master_binlog_checksum"); - } catch (IOException e) { - throw new CanalParseException(e); - } - - List columnValues = rs.getFieldValues(); - if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0) != null - && columnValues.get(0).toUpperCase().equals("CRC32")) { - binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32; - } else { - binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF; - } - } - public static enum BinlogFormat { STATEMENT("STATEMENT"), ROW("ROW"), MIXED("MIXED"); From 9ba93bae5f2fc6ad51add52c36d5588ba1511712 Mon Sep 17 00:00:00 2001 From: luoyaogui Date: Tue, 27 Jun 2017 15:08:34 +0800 Subject: [PATCH 9/9] Update DirectLogFetcher.java --- .../mysql/dbsync/DirectLogFetcher.java | 19 +++---------------- 1 file changed, 3 insertions(+), 16 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java index 6ad2e5db..dc0b1b1a 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java @@ -3,13 +3,12 @@ package com.alibaba.otter.canal.parse.inbound.mysql.dbsync; import java.io.IOException; import java.io.InterruptedIOException; import java.net.SocketTimeoutException; -import java.nio.ByteBuffer; import java.nio.channels.ClosedByInterruptException; +import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; import com.taobao.tddl.dbsync.binlog.LogFetcher; /** @@ -149,20 +148,8 @@ public class DirectLogFetcher extends LogFetcher { private final boolean fetch0(final int off, final int len) throws IOException { ensureCapacity(off + len); - ByteBuffer buffer = ByteBuffer.wrap(this.buffer, off, len); - while (buffer.hasRemaining()) { - int readNum = channel.read(buffer); - if (readNum == -1) { - throw new IOException("Unexpected End Stream"); - } - } - - // for (int count, n = 0; n < len; n += count) { - // if (0 > (count = input.read(buffer, off + n, len - n))) { - // // Reached end of input stream - // return false; - // } - // } + byte[] read = channel.read(len); + System.arraycopy(read, 0, this.buffer, off, len); if (limit < off + len) limit = off + len; return true;