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

fixed socket wait period 10ms

Browse Source
This commit is contained in:
agapple
2018-02-12 11:43:03 +08:00
parent 826a45c105
commit 2d49b709ca
3 changed files with 12 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java
View File

@@ -43,8 +43,8 @@ public class MysqlConnector {
// mysql connectinnId
private long connectionId = -1;
private AtomicBoolean connected = new AtomicBoolean(false);
public static final int timeout = 3000; // 3s
public static final int timeout = 5 * 1000; // 5s
public MysqlConnector(){
}

15
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java
View File

@@ -15,6 +15,7 @@ import java.net.SocketAddress;
*/
public class SocketChannel {
private static final int period = 10;
private Channel channel = null;
private Object lock = new Object();
private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(1024 * 1024); // 缓存大小
@@ -31,7 +32,7 @@ public class SocketChannel {
synchronized (lock) {
while (true) {
cache.discardReadBytes();// 回收内存
//source buffer is empty.
// source buffer is empty.
if (!buf.isReadable()) {
break;
}
@@ -39,8 +40,8 @@ public class SocketChannel {
if (cache.isWritable()) {
cache.writeBytes(buf, Math.min(cache.writableBytes(), buf.readableBytes()));
} else {
//dest buffer is full.
lock.wait(100);
// dest buffer is full.
lock.wait(period);
}
}
}
@@ -62,7 +63,7 @@ public class SocketChannel {
}
synchronized (this) {
try {
wait(100);
wait(period);
} catch (InterruptedException e) {
throw new java.nio.channels.ClosedByInterruptException();
}
@@ -76,7 +77,7 @@ public class SocketChannel {
}
} while (true);
}
public byte[] read(int readSize, int timeout) throws IOException {
int accumulatedWaitTime = 0;
do {
@@ -85,14 +86,14 @@ public class SocketChannel {
throw new IOException("socket has Interrupted !");
}
accumulatedWaitTime += 100;
accumulatedWaitTime += period;
if (accumulatedWaitTime > timeout) {
throw new IOException("socket read timeout occured !");
}
synchronized (this) {
try {
wait(100);
wait(period);
} catch (InterruptedException e) {
throw new IOException("socket has Interrupted !");
}

5
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java
View File

@@ -7,13 +7,12 @@ import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.ReferenceCountUtil;
import java.io.IOException;
import java.net.SocketAddress;
@@ -103,7 +102,7 @@ public abstract class SocketChannelPool {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
//need output error for troubeshooting.
// need output error for troubeshooting.
logger.error("business error.", cause);
ctx.close();
}