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 a01b8b6a..898ed5eb 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 @@ -10,7 +10,9 @@ import java.net.SocketAddress; import java.nio.ByteBuffer; /** - * @author luoyaogui 封装netty的通信channel和数据接收缓存,实现读、写、连接校验的功能。 2016-12-28 + * 封装netty的通信channel和数据接收缓存,实现读、写、连接校验的功能。 2016-12-28 + * + * @author luoyaogui */ public class SocketChannel { @@ -24,7 +26,7 @@ public class SocketChannel { public void setChannel(Channel channel, boolean notify) { this.channel = channel; - if (notify) {// 是否需要通知,主要时channel不可用时 + if (notify) {// 是否需要通知,主要是channel不可用时 synchronized (this) { notifyAll(); } @@ -50,12 +52,15 @@ public class SocketChannel { } public int read(ByteBuffer buffer) throws IOException { - if (null == channel) throw new IOException("socket has Interrupted !"); + if (null == channel) { + throw new IOException("socket has Interrupted !"); + } if (cache.readableBytes() < buffer.remaining()) { synchronized (this) { try { wait(); } catch (InterruptedException e) { + throw new IOException("socket has Interrupted !"); } } } else { 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 0d219346..68990903 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 @@ -24,7 +24,7 @@ import java.util.concurrent.ConcurrentHashMap; /** * @author luoyaogui 实现channel的管理(监听连接、读数据、回收) 2016-12-28 */ -@SuppressWarnings("rawtypes") +@SuppressWarnings({ "rawtypes", "deprecation" }) public abstract class SocketChannelPool { private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组 @@ -57,7 +57,9 @@ public abstract class SocketChannelPool { @Override public void operationComplete(ChannelFuture arg0) throws Exception { - if (arg0.isSuccess()) socket.setChannel(arg0.channel(), false); + if (arg0.isSuccess()) { + socket.setChannel(arg0.channel(), false); + } synchronized (socket) { socket.notify(); }