From 8ad603fea471e6a45b2659f12ac17fa306cca4ed Mon Sep 17 00:00:00 2001 From: Wu-Jianqiang <959007@163.com> Date: Mon, 26 Feb 2018 16:25:31 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9ASocketChannel=20?= =?UTF-8?q?=E9=BB=98=E8=AE=A4=E7=BC=93=E5=AD=98=E5=A4=A7=E5=B0=8F=EF=BC=88?= =?UTF-8?q?1MB=EF=BC=89=E4=B8=8D=E5=A4=9F=E7=94=A8=E6=97=B6=E9=9C=80?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E6=89=A9=E5=85=85=EF=BC=8C=E5=90=A6=E5=88=99?= =?UTF-8?q?=E5=B0=86=E5=9B=A0=E7=BC=93=E5=AD=98=E7=A9=BA=E9=97=B4=E4=B8=8D?= =?UTF-8?q?=E8=B6=B3=E8=80=8C=E9=80=A0=E6=88=90I/O=E8=B6=85=E6=97=B6?= =?UTF-8?q?=E5=81=87=E8=B1=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../driver/mysql/socket/SocketChannel.java | 71 +++++++++++-------- 1 file changed, 43 insertions(+), 28 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 b55b6722..968e0eeb 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 @@ -15,10 +15,12 @@ import java.net.SocketAddress; */ public class SocketChannel { - private static final int period = 10; + private static final int WAIT_PERIOD = 10; // milliseconds + private static final int DEFAULT_INIT_BUFFER_SIZE = 1024 * 1024; // 1MB，默认初始缓存大小 + private static final int DEFAULT_MAX_BUFFER_SIZE = 4 * DEFAULT_INIT_BUFFER_SIZE; // 4MB，默认最大缓存大小 private Channel channel = null; private Object lock = new Object(); - private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(1024 * 1024); // 缓存大小 + private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(DEFAULT_INIT_BUFFER_SIZE); // 缓存大小 public Channel getChannel() { return channel; @@ -28,9 +30,13 @@ public class SocketChannel { this.channel = channel; } - public void writeCache(ByteBuf buf) throws InterruptedException { + public void writeCache(ByteBuf buf) throws InterruptedException, IOException { synchronized (lock) { while (true) { + if (null == cache) { + throw new IOException("socket is closed !"); + } + cache.discardReadBytes();// 回收内存 // source buffer is empty. if (!buf.isReadable()) { @@ -41,7 +47,7 @@ public class SocketChannel { cache.writeBytes(buf, Math.min(cache.writableBytes(), buf.readableBytes())); } else { // dest buffer is full. - lock.wait(period); + lock.wait(WAIT_PERIOD); } } } @@ -56,44 +62,46 @@ public class SocketChannel { } public byte[] read(int readSize) throws IOException { - do { - if (readSize > cache.readableBytes()) { - if (null == channel) { - throw new java.nio.channels.ClosedByInterruptException(); - } - synchronized (this) { - try { - wait(period); - } catch (InterruptedException e) { - throw new java.nio.channels.ClosedByInterruptException(); - } - } - } else { - byte[] back = new byte[readSize]; - synchronized (lock) { - cache.readBytes(back); - } - return back; - } - } while (true); + return read(readSize, 0); } public byte[] read(int readSize, int timeout) throws IOException { int accumulatedWaitTime = 0; + + // 若读取内容较长，则自动扩充超时时间，以初始缓存大小为基准计算倍数 + if (timeout > 0 && readSize > DEFAULT_INIT_BUFFER_SIZE ) { + timeout *= (readSize / DEFAULT_INIT_BUFFER_SIZE + 1); + } do { if (readSize > cache.readableBytes()) { if (null == channel) { throw new IOException("socket has Interrupted !"); } - accumulatedWaitTime += period; - if (accumulatedWaitTime > timeout) { - throw new IOException("socket read timeout occured !"); + // 默认缓存大小不够用时需自动扩充，否则将因缓存空间不足而造成I/O超时假象 + if (!cache.isWritable(readSize - cache.readableBytes())) { + synchronized (lock) { + int deltaSize = readSize - cache.readableBytes(); // 同步锁后重新读取 + deltaSize = deltaSize - cache.writableBytes(); + if (deltaSize > 0) { + deltaSize = (deltaSize / 32 + 1) * 32; + cache.capacity(cache.capacity() + deltaSize); + } + } + } else if (timeout > 0) { + accumulatedWaitTime += WAIT_PERIOD; + if (accumulatedWaitTime > timeout) { + StringBuilder sb = new StringBuilder("socket read timeout occured !"); + sb.append(" readSize = ").append(readSize); + sb.append(", readableBytes = ").append(cache.readableBytes()); + sb.append(", timeout = ").append(timeout); + throw new IOException(sb.toString()); + } } synchronized (this) { try { - wait(period); + wait(WAIT_PERIOD); } catch (InterruptedException e) { throw new IOException("socket has Interrupted !"); } @@ -102,6 +110,13 @@ public class SocketChannel { byte[] back = new byte[readSize]; synchronized (lock) { cache.readBytes(back); + // 恢复自动扩充的过大缓存到默认初始大小，释放空间 + if (cache.capacity() > DEFAULT_MAX_BUFFER_SIZE) { + cache.discardReadBytes(); // 回收已读空间，重置读写指针 + if (cache.readableBytes() < DEFAULT_INIT_BUFFER_SIZE) { + cache.capacity(DEFAULT_INIT_BUFFER_SIZE); + } + } } return back; }