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 1/3] =?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; } From 06e7f441144247275a51561fa7eb6ab2a616405c Mon Sep 17 00:00:00 2001 From: Wu-Jianqiang <959007@163.com> Date: Mon, 26 Feb 2018 16:31:58 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9AMASTER=5FHEARTBE?= =?UTF-8?q?AT=5FPERIOD=20=E7=9A=84=E8=AE=A1=E9=87=8F=E5=8D=95=E4=BD=8D?= =?UTF-8?q?=E6=98=AF=E7=A7=92=EF=BC=88=E5=8F=AF=E4=BB=A5=E5=8C=85=E5=90=AB?= =?UTF-8?q?3=E4=BD=8D=E5=B0=8F=E6=95=B0=E8=A1=A8=E7=A4=BA=E6=AF=AB?= =?UTF-8?q?=E7=A7=92=EF=BC=89=EF=BC=8C=E4=BD=86=E4=B8=8D=E6=98=AF=E7=BA=B3?= =?UTF-8?q?=E7=A7=92=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/parse/inbound/mysql/MysqlConnection.java | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 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 e665b4f5..0eeef9f5 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 @@ -297,9 +297,15 @@ public class MysqlConnection implements ErosaConnection { logger.warn("update mariadb_slave_capability failed", e); } - long periodNano = TimeUnit.SECONDS.toNanos(MASTER_HEARTBEAT_PERIOD_SECONDS); + /** + * MASTER_HEARTBEAT_PERIOD sets the interval in seconds between replication heartbeats. + * Whenever the master's binary log is updated with an event, the waiting period for the next heartbeat is reset. + * interval is a decimal value having the range 0 to 4294967 seconds and a resolution in milliseconds; + * the smallest nonzero value is 0.001. Heartbeats are sent by the master only if there are no unsent events + * in the binary log file for a period longer than interval. + */ try { - update("SET @master_heartbeat_period=" + periodNano); + update("SET @master_heartbeat_period=" + MASTER_HEARTBEAT_PERIOD_SECONDS); } catch (Exception e) { logger.warn("update master_heartbeat_period failed", e); } From 3e7e7b7f6e0cd9a525ecf96b3192e7f47d262d15 Mon Sep 17 00:00:00 2001 From: Wu-Jianqiang <959007@163.com> Date: Mon, 26 Feb 2018 16:37:21 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=E4=BF=AE=E6=94=B9=EF=BC=9A=E5=8F=98?= =?UTF-8?q?=E9=87=8F=E5=91=BD=E5=90=8D=E5=92=8Csuper=E8=B0=83=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/parse/driver/mysql/socket/SocketChannelPool.java | 5 +++-- 1 file changed, 3 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 fa16a829..efec72c6 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 @@ -48,8 +48,8 @@ public abstract class SocketChannelPool { .handler(new ChannelInitializer() { @Override - protected void initChannel(Channel arg0) throws Exception { - arg0.pipeline().addLast(new BusinessHandler());// 命令过滤和handler添加管理 + protected void initChannel(Channel ch) throws Exception { + ch.pipeline().addLast(new BusinessHandler());// 命令过滤和handler添加管理 } }); } @@ -79,6 +79,7 @@ public abstract class SocketChannelPool { public void channelInactive(ChannelHandlerContext ctx) throws Exception { socket.setChannel(null); chManager.remove(ctx.channel());// 移除 + super.channelInactive(ctx); } @Override