From 788c1d9b84e470d28eaf1198ad0a5d7f18da287b Mon Sep 17 00:00:00 2001 From: agapple Date: Sat, 23 Sep 2017 11:50:25 +0800 Subject: [PATCH] fixed issue #360 , socket pool bug --- .../mysql/socket/SocketChannelPool.java | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 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 13efbed9..52609b19 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 @@ -21,14 +21,16 @@ import java.net.SocketAddress; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import com.alibaba.otter.canal.common.utils.BooleanMutex; + /** * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28 */ @SuppressWarnings({ "rawtypes", "deprecation" }) public abstract class SocketChannelPool { - private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组 - private static Bootstrap boot = new Bootstrap(); // 主 + private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组 + private static Bootstrap boot = new Bootstrap(); // 主 private static Map chManager = new ConcurrentHashMap(); static { @@ -53,19 +55,20 @@ public abstract class SocketChannelPool { public static SocketChannel open(SocketAddress address) throws Exception { final SocketChannel socket = new SocketChannel(); + final BooleanMutex mutex = new BooleanMutex(false); boot.connect(address).addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture arg0) throws Exception { - if (arg0.isSuccess()) socket.setChannel(arg0.channel()); - synchronized (socket) { - socket.notify(); + if (arg0.isSuccess()) { + socket.setChannel(arg0.channel()); } + + mutex.set(true); } }); - synchronized (socket) { - socket.wait(); - } + // wait for complete + mutex.get(); if (null == socket.getChannel()) { throw new IOException("can't create socket!"); }