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 issue #360 , socket pool bug

Browse Source
This commit is contained in:
agapple
2017-09-23 11:50:25 +08:00
parent 84a9aeb46c
commit 788c1d9b84
1 changed files with 11 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,14 +21,16 @@ import java.net.SocketAddress;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import com.alibaba.otter.canal.common.utils.BooleanMutex;
/** /**
* @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28 * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
*/ */
@SuppressWarnings({ "rawtypes", "deprecation" }) @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线程组
private static Bootstrap boot = new Bootstrap(); // 主 private static Bootstrap boot = new Bootstrap(); // 主
private static Map<Channel, SocketChannel> chManager = new ConcurrentHashMap<Channel, SocketChannel>(); private static Map<Channel, SocketChannel> chManager = new ConcurrentHashMap<Channel, SocketChannel>();
static { static {
@@ -53,19 +55,20 @@ public abstract class SocketChannelPool {
public static SocketChannel open(SocketAddress address) throws Exception { public static SocketChannel open(SocketAddress address) throws Exception {
final SocketChannel socket = new SocketChannel(); final SocketChannel socket = new SocketChannel();
final BooleanMutex mutex = new BooleanMutex(false);
boot.connect(address).addListener(new ChannelFutureListener() { boot.connect(address).addListener(new ChannelFutureListener() {
@Override @Override
public void operationComplete(ChannelFuture arg0) throws Exception { public void operationComplete(ChannelFuture arg0) throws Exception {
if (arg0.isSuccess()) socket.setChannel(arg0.channel()); if (arg0.isSuccess()) {
synchronized (socket) { socket.setChannel(arg0.channel());
socket.notify();
} }
mutex.set(true);
} }
}); });
synchronized (socket) { // wait for complete
socket.wait(); mutex.get();
}
if (null == socket.getChannel()) { if (null == socket.getChannel()) {
throw new IOException("can't create socket!"); throw new IOException("can't create socket!");
} }