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 stop failed

Browse Source
This commit is contained in:
agapple
2017-01-19 13:22:40 +08:00
parent ff97dcbbc9
commit 0ee2e767d2
2 changed files with 12 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,7 +10,9 @@ import java.net.SocketAddress;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
/** /**
* @author luoyaogui 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。 2016-12-28 * 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。 2016-12-28
*
* @author luoyaogui
*/ */
public class SocketChannel { public class SocketChannel {
@@ -24,7 +26,7 @@ public class SocketChannel {
public void setChannel(Channel channel, boolean notify) { public void setChannel(Channel channel, boolean notify) {
this.channel = channel; this.channel = channel;
if (notify) {// 是否需要通知，主要时channel不可用时 if (notify) {// 是否需要通知，主要是channel不可用时
synchronized (this) { synchronized (this) {
notifyAll(); notifyAll();
} }
@@ -50,12 +52,15 @@ public class SocketChannel {
} }
public int read(ByteBuffer buffer) throws IOException { 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()) { if (cache.readableBytes() < buffer.remaining()) {
synchronized (this) { synchronized (this) {
try { try {
wait(); wait();
} catch (InterruptedException e) { } catch (InterruptedException e) {
throw new IOException("socket has Interrupted !");
} }
} }
} else { } else {

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

@@ -24,7 +24,7 @@ import java.util.concurrent.ConcurrentHashMap;
/** /**
* @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28 * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
*/ */
@SuppressWarnings("rawtypes") @SuppressWarnings({ "rawtypes", "deprecation" })
public abstract class SocketChannelPool { public abstract class SocketChannelPool {
private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组 private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组
@@ -57,7 +57,9 @@ public abstract class SocketChannelPool {
@Override @Override
public void operationComplete(ChannelFuture arg0) throws Exception { 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) { synchronized (socket) {
socket.notify(); socket.notify();
} }