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

Update SocketChannel.java

Browse Source
This commit is contained in:
luoyaogui
2017-06-27 15:02:45 +08:00
committed by GitHub
parent 36216e63ea
commit 48c108806a
1 changed files with 17 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -7,7 +7,6 @@ import io.netty.channel.Channel;
import java.io.IOException; import java.io.IOException;
import java.net.SocketAddress; import java.net.SocketAddress;
import java.nio.ByteBuffer;
/** /**
* @author luoyaogui * @author luoyaogui
@@ -17,18 +16,13 @@ import java.nio.ByteBuffer;
public class SocketChannel { public class SocketChannel {
private Channel channel = null; private Channel channel = null;
private Object lock = new Object(); 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() { public Channel getChannel() {
return channel; return channel;
} }
public void setChannel(Channel channel,boolean notify) { public void setChannel(Channel channel) {
this.channel = channel; this.channel = channel;
if(notify){//是否需要通知，主要时channel不可用时
synchronized(this){
notifyAll();
}
}
} }
public void writeCache(ByteBuf buf){ public void writeCache(ByteBuf buf){
@@ -36,9 +30,6 @@ public class SocketChannel {
cache.discardReadBytes();//回收内存 cache.discardReadBytes();//回收内存
cache.writeBytes(buf); cache.writeBytes(buf);
} }
synchronized(this){
notifyAll();
}
} }
public void writeChannel(byte[]... buf) throws IOException { public void writeChannel(byte[]... buf) throws IOException {
if(channel != null && channel.isWritable()) if(channel != null && channel.isWritable())
@@ -46,19 +37,22 @@ public class SocketChannel {
else else
throw new IOException("write failed ! please checking !"); throw new IOException("write failed ! please checking !");
} }
public int read(ByteBuffer buffer) throws IOException { public byte[] read(int readSize) throws IOException {
do{
if(readSize > cache.readableBytes()){
if(null == channel) if(null == channel)
throw new IOException("socket has Interrupted !"); throw new IOException("socket has Interrupted !");
if(cache.readableBytes() < buffer.remaining()){
synchronized (this) { synchronized (this) {
try {wait();} catch (InterruptedException e) {} try { wait(100); } catch (InterruptedException e) {}
} }
} else { } else {
byte[] back = new byte[readSize];
synchronized (lock) { synchronized (lock) {
cache.readBytes(buffer); cache.readBytes(back);
} }
return back;
} }
return 0; }while(true);
} }
public boolean isConnected() { public boolean isConnected() {
return channel!=null?true:false; return channel!=null?true:false;