From 472e91a4e98383bfdf3dcd8dc5528c942b497135 Mon Sep 17 00:00:00 2001 From: "chao.liuc" Date: Tue, 25 Oct 2011 09:18:22 +0000 Subject: [PATCH] =?UTF-8?q?DUBBO-508=20remoting:=E6=8A=8A=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E7=9A=84listener=E6=94=B9=E6=88=90await?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit git-svn-id: http://code.alibabatech.com/svn/dubbo/trunk@44 1a56cb94-b969-4eaa-88fa-be21384802f2 --- .../transport/support/netty/NettyClient.java | 165 ++++++++---------- 1 file changed, 74 insertions(+), 91 deletions(-) diff --git a/dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/transport/support/netty/NettyClient.java b/dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/transport/support/netty/NettyClient.java index 610d100500..5f8d70b4e1 100644 --- a/dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/transport/support/netty/NettyClient.java +++ b/dubbo-remoting/src/main/java/com/alibaba/dubbo/remoting/transport/support/netty/NettyClient.java @@ -15,30 +15,27 @@ */ package com.alibaba.dubbo.remoting.transport.support.netty; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; - -import org.jboss.netty.bootstrap.ClientBootstrap; -import org.jboss.netty.channel.Channel; -import org.jboss.netty.channel.ChannelFactory; -import org.jboss.netty.channel.ChannelFuture; -import org.jboss.netty.channel.ChannelFutureListener; -import org.jboss.netty.channel.ChannelPipeline; -import org.jboss.netty.channel.ChannelPipelineFactory; -import org.jboss.netty.channel.Channels; -import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory; - -import com.alibaba.dubbo.common.Constants; -import com.alibaba.dubbo.common.URL; -import com.alibaba.dubbo.common.Version; -import com.alibaba.dubbo.common.utils.NamedThreadFactory; -import com.alibaba.dubbo.common.utils.NetUtils; -import com.alibaba.dubbo.remoting.ChannelHandler; -import com.alibaba.dubbo.remoting.RemotingException; -import com.alibaba.dubbo.remoting.transport.support.AbstractClient; -import com.alibaba.dubbo.remoting.transport.support.handler.ChannelHandlers; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.jboss.netty.bootstrap.ClientBootstrap; +import org.jboss.netty.channel.Channel; +import org.jboss.netty.channel.ChannelFactory; +import org.jboss.netty.channel.ChannelFuture; +import org.jboss.netty.channel.ChannelPipeline; +import org.jboss.netty.channel.ChannelPipelineFactory; +import org.jboss.netty.channel.Channels; +import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory; + +import com.alibaba.dubbo.common.Constants; +import com.alibaba.dubbo.common.URL; +import com.alibaba.dubbo.common.Version; +import com.alibaba.dubbo.common.utils.NamedThreadFactory; +import com.alibaba.dubbo.common.utils.NetUtils; +import com.alibaba.dubbo.remoting.ChannelHandler; +import com.alibaba.dubbo.remoting.RemotingException; +import com.alibaba.dubbo.remoting.transport.support.AbstractClient; +import com.alibaba.dubbo.remoting.transport.support.handler.ChannelHandlers; /** * NettyClient. @@ -86,73 +83,59 @@ public class NettyClient extends AbstractClient { }); } - protected void doConnect() throws Throwable { - ChannelFuture future = bootstrap.connect(getConnectAddress()); - try { - long start = System.currentTimeMillis(); - final AtomicReference exception = new AtomicReference(); - final CountDownLatch finish = new CountDownLatch(1); // resolve future.awaitUninterruptibly() dead lock - future.addListener(new ChannelFutureListener() { - public void operationComplete(ChannelFuture future) throws Exception { - try { - if (future.isSuccess()) { - Channel newChannel = future.getChannel(); - newChannel.setInterestOps(Channel.OP_READ_WRITE); - try { - // 关闭旧的连接 - Channel oldChannel = NettyClient.this.channel; // copy reference - if (oldChannel != null) { - try { - if (logger.isInfoEnabled()) { - logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel); - } - oldChannel.close(); - } finally { - NettyChannel.removeChannelIfDisconnected(oldChannel); - } - } - } finally { - if (NettyClient.this.isClosed()) { - try { - if (logger.isInfoEnabled()) { - logger.info("Close new netty channel " + newChannel + ", because the client closed."); - } - newChannel.close(); - } finally { - NettyClient.this.channel = null; - NettyChannel.removeChannelIfDisconnected(newChannel); - } - } else { - NettyClient.this.channel = newChannel; - } - } - } else if (future.getCause() != null) { - exception.set(future.getCause()); - } - } catch (Exception e) { - exception.set(e); - } finally { - finish.countDown(); - } - } - }); - try { - finish.await(getConnectTimeout(), TimeUnit.MILLISECONDS); - } catch (InterruptedException e) { - throw new RemotingException(this, "Failed to connect to server " + getRemoteAddress() + " client-side timeout " - + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) - + "ms) from netty client " + NetUtils.getLocalHost() + " using dubbo version " - + Version.getVersion() + ", cause: " + e.getMessage(), e); - } - Throwable e = exception.get(); - if (e != null) { - throw e; - } - } finally { - if (! isConnected()) { - future.cancel(); - } - } + protected void doConnect() throws Throwable { + long start = System.currentTimeMillis(); + ChannelFuture future = bootstrap.connect(getConnectAddress()); + try{ + boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS); + + if (ret && future.isSuccess()) { + Channel newChannel = future.getChannel(); + newChannel.setInterestOps(Channel.OP_READ_WRITE); + try { + // 关闭旧的连接 + Channel oldChannel = NettyClient.this.channel; // copy reference + if (oldChannel != null) { + try { + if (logger.isInfoEnabled()) { + logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel); + } + oldChannel.close(); + } finally { + NettyChannel.removeChannelIfDisconnected(oldChannel); + } + } + } finally { + if (NettyClient.this.isClosed()) { + try { + if (logger.isInfoEnabled()) { + logger.info("Close new netty channel " + newChannel + ", because the client closed."); + } + newChannel.close(); + } finally { + NettyClient.this.channel = null; + NettyChannel.removeChannelIfDisconnected(newChannel); + } + } else { + NettyClient.this.channel = newChannel; + } + } + } else if (future.getCause() != null) { + throw future.getCause(); + } else { + throw new RemotingException(this, "Faild to connect to server " + getRemoteAddress() + ", the future was not completed within the specified time limit, please check the timeout ["+ getConnectTimeout() +"] config ."); + } + + }catch (InterruptedException e) { + throw new RemotingException(this, "Failed to connect to server " + getRemoteAddress() + " client-side timeout " + + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + + "ms) from netty client " + NetUtils.getLocalHost() + " using dubbo version " + + Version.getVersion() + ", cause: " + e.getMessage(), e); + }finally{ + if (! isConnected()) { + future.cancel(); + } + } } @Override