dubbo網絡通信(四)
dubbo網絡層功能
dubbo網絡通信層主要實現瞭如下功能：java
- 多種網絡通信框架的抽象封裝（netty,mina,grizzly）
- 每一個客戶端主機和服務端保存單個長連接通訊
- 異步調用轉同步
- tcp連接的心跳和自動重連
- 基於header頭通信協議，請求的編解碼器
dubbo網絡通信框架抽象
dubbo的網絡通訊基於NIO框架，通常基於事件的NIO網絡框架都涉及到 channel , channelHandle核心概念，網絡數據buffer, 網絡數據編解碼器，dubbo爲了可以適配多種NIO框架，將以上概念所有又抽象了一層接口。若是有netty開發經驗或者瞭解netty helloworld demo程序對於理解這個章節很是有幫助。網絡
/* *如下是dubbo對於channel , channelHandle封裝的抽象接口 */
public interface Endpoint {
/** * get url. * * @return url */
URL getUrl();
/** * get channel handler. * * @return channel handler */
ChannelHandler getChannelHandler();
/** * get local address. * * @return local address. */
InetSocketAddress getLocalAddress();
/** * send message. * * @param message * @throws RemotingException */
void send(Object message) throws RemotingException;
/** * send message. * * @param message * @param sent 是否已發送完成 */
void send(Object message, boolean sent) throws RemotingException;
/** * close the channel. */
void close();
/** * Graceful close the channel. */
void close(int timeout);
/** * is closed. * * @return closed */
boolean isClosed();
}
public interface Channel extends Endpoint {
/** * get remote address. * * @return remote address. */
InetSocketAddress getRemoteAddress();
/** * is connected. * * @return connected */
boolean isConnected();
/** * has attribute. * * @param key key. * @return has or has not. */
boolean hasAttribute(String key);
/** * get attribute. * * @param key key. * @return value. */
Object getAttribute(String key);
/** * set attribute. * * @param key key. * @param value value. */
void setAttribute(String key,Object value);
/** * remove attribute. * * @param key key. */
void removeAttribute(String key);
}
public interface Client extends Endpoint, Channel, Resetable {
/** * reconnect. */
void reconnect() throws RemotingException;
@Deprecated
void reset(com.alibaba.dubbo.common.Parameters parameters);
}
@SPI
public interface ChannelHandler {
/** * on channel connected. * * @param channel channel. */
void connected(Channel channel) throws RemotingException;
/** * on channel disconnected. * * @param channel channel. */
void disconnected(Channel channel) throws RemotingException;
/** * on message sent. * * @param channel channel. * @param message message. */
void sent(Channel channel, Object message) throws RemotingException;
/** * on message received.接收到對方消息和回調 * * @param channel channel. * @param message message. */
void received(Channel channel, Object message) throws RemotingException;
/** * on exception caught. * * @param channel channel. * @param exception exception. */
void caught(Channel channel, Throwable exception) throws RemotingException;
}
/* *信息交換client，對Client對象接口包裝，實現發送請求返回結果Future 以及長連接心跳定時發送 */
public class HeaderExchangeClient implements ExchangeClient {
private static final Logger logger = LoggerFactory.getLogger( HeaderExchangeClient.class );
private static final ScheduledThreadPoolExecutor scheduled = new ScheduledThreadPoolExecutor(2, new NamedThreadFactory("dubbo-remoting-client-heartbeat", true));
// 心跳定時器
private ScheduledFuture<?> heatbeatTimer;
// 心跳超時，毫秒。缺省0，不會執行心跳。
private int heartbeat;
private int heartbeatTimeout;
private final Client client;
private final ExchangeChannel channel;
public HeaderExchangeClient(Client client){
if (client == null) {
throw new IllegalArgumentException("client == null");
}
this.client = client;
this.channel = new HeaderExchangeChannel(client);
String dubbo = client.getUrl().getParameter(Constants.DUBBO_VERSION_KEY);
this.heartbeat = client.getUrl().getParameter( Constants.HEARTBEAT_KEY, dubbo != null && dubbo.startsWith("1.0.") ? Constants.DEFAULT_HEARTBEAT : 0 );
this.heartbeatTimeout = client.getUrl().getParameter( Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3 );
if ( heartbeatTimeout < heartbeat * 2 ) {
throw new IllegalStateException( "heartbeatTimeout < heartbeatInterval * 2" );
}
startHeatbeatTimer();
}
//最核心方法，實現發送請求後返回ResponseFuture對象，經過該Future能夠獲取RPC請求結果
public ResponseFuture request(Object request) throws RemotingException {
return channel.request(request);
}
public URL getUrl() {
return channel.getUrl();
}
public InetSocketAddress getRemoteAddress() {
return channel.getRemoteAddress();
}
public ResponseFuture request(Object request, int timeout) throws RemotingException {
return channel.request(request, timeout);
}
public ChannelHandler getChannelHandler() {
return channel.getChannelHandler();
}
public boolean isConnected() {
return channel.isConnected();
}
public InetSocketAddress getLocalAddress() {
return channel.getLocalAddress();
}
public ExchangeHandler getExchangeHandler() {
return channel.getExchangeHandler();
}
public void send(Object message) throws RemotingException {
channel.send(message);
}
public void send(Object message, boolean sent) throws RemotingException {
channel.send(message, sent);
}
public boolean isClosed() {
return channel.isClosed();
}
public void close() {
doClose();
channel.close();
}
public void close(int timeout) {
doClose();
channel.close(timeout);
}
public void reset(URL url) {
client.reset(url);
}
@Deprecated
public void reset(com.alibaba.dubbo.common.Parameters parameters){
reset(getUrl().addParameters(parameters.getParameters()));
}
public void reconnect() throws RemotingException {
client.reconnect();
}
public Object getAttribute(String key) {
return channel.getAttribute(key);
}
public void setAttribute(String key, Object value) {
channel.setAttribute(key, value);
}
public void removeAttribute(String key) {
channel.removeAttribute(key);
}
public boolean hasAttribute(String key) {
return channel.hasAttribute(key);
}
private void startHeatbeatTimer() {
stopHeartbeatTimer();
if ( heartbeat > 0 ) {
heatbeatTimer = scheduled.scheduleWithFixedDelay(
new HeartBeatTask( new HeartBeatTask.ChannelProvider() {
public Collection<Channel> getChannels() {
return Collections.<Channel>singletonList( HeaderExchangeClient.this );
}
}, heartbeat, heartbeatTimeout),
heartbeat, heartbeat, TimeUnit.MILLISECONDS );
}
}
private void stopHeartbeatTimer() {
if (heatbeatTimer != null && ! heatbeatTimer.isCancelled()) {
try {
heatbeatTimer.cancel(true);
scheduled.purge();
} catch ( Throwable e ) {
if (logger.isWarnEnabled()) {
logger.warn(e.getMessage(), e);
}
}
}
heatbeatTimer =null;
}
private void doClose() {
stopHeartbeatTimer();
}
@Override
public String toString() {
return "HeaderExchangeClient [channel=" + channel + "]";
}
}
/* *基於Header Request信息交換Channel，對於真實底層Chanel實現好比Netty *Channel進行了一層包裝，實現將RPC請求參數Invocation封裝爲Request對象, *Request對象描述了RPC請求所有信息。協議頭數據以及協議body數據。 */
final class HeaderExchangeChannel implements ExchangeChannel {
private static final Logger logger = LoggerFactory.getLogger(HeaderExchangeChannel.class);
private static final String CHANNEL_KEY = HeaderExchangeChannel.class.getName() + ".CHANNEL";
private final Channel channel;
private volatile boolean closed = false;
HeaderExchangeChannel(Channel channel){
if (channel == null) {
throw new IllegalArgumentException("channel == null");
}
this.channel = channel;
}
static HeaderExchangeChannel getOrAddChannel(Channel ch) {
if (ch == null) {
return null;
}
HeaderExchangeChannel ret = (HeaderExchangeChannel) ch.getAttribute(CHANNEL_KEY);
if (ret == null) {
ret = new HeaderExchangeChannel(ch);
if (ch.isConnected()) {
ch.setAttribute(CHANNEL_KEY, ret);
}
}
return ret;
}
static void removeChannelIfDisconnected(Channel ch) {
if (ch != null && ! ch.isConnected()) {
ch.removeAttribute(CHANNEL_KEY);
}
}
public void send(Object message) throws RemotingException {
send(message, getUrl().getParameter(Constants.SENT_KEY, false));
}
public void send(Object message, boolean sent) throws RemotingException {
if (closed) {
throw new RemotingException(this.getLocalAddress(), null, "Failed to send message " + message + ", cause: The channel " + this + " is closed!");
}
if (message instanceof Request
|| message instanceof Response
|| message instanceof String) {
channel.send(message, sent);
} else {
Request request = new Request();
request.setVersion("2.0.0");
request.setTwoWay(false);
request.setData(message);
channel.send(request, sent);
}
}
public ResponseFuture request(Object request) throws RemotingException {
return request(request, channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
}
//request 入參類型是Invocation invocation
public ResponseFuture request(Object request, int timeout) throws RemotingException {
if (closed) {
throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
}
// create request.將Invocation封裝爲Request對象，request對象建立的時候會生成遞增的請求id,private static final AtomicLong INVOKE_ID = new AtomicLong(0);
Request req = new Request();
req.setVersion("2.0.0");
req.setTwoWay(true);
req.setData(request);
//發送請求數據是異步的，返回future對象，調用方用於同步等待響應，相似於提交到線程池返回Future對象用於等待返回結果
DefaultFuture future = new DefaultFuture(channel, req, timeout);
try{
channel.send(req);
}catch (RemotingException e) {
future.cancel();
throw e;
}
return future;
}
public class DefaultFuture implements ResponseFuture {
//存儲全部requestId 和 通道映射，用於判斷當前通道是否有未完成的請求
private static final Map<Long, Channel> CHANNELS = new ConcurrentHashMap<Long, Channel>();
//存儲請求id和future對象映射，用於收到響應喚醒等待線程
private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();
// invoke id.
private final long id;
private final Channel channel;
private final Request request;
private final int timeout;
private final Lock lock = new ReentrantLock();
private final Condition done = lock.newCondition();
private final long start = System.currentTimeMillis();
private volatile long sent;
private volatile Response response;
private volatile ResponseCallback callback;
public DefaultFuture(Channel channel, Request request, int timeout){
this.channel = channel;
this.request = request;
this.id = request.getId();
this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
// put into waiting map.
FUTURES.put(id, this);
CHANNELS.put(id, channel);
}
public Object get() throws RemotingException {
return get(timeout);
}
public Object get(int timeout) throws RemotingException {
if (timeout <= 0) {
timeout = Constants.DEFAULT_TIMEOUT;
}
if (! isDone()) {
long start = System.currentTimeMillis();
lock.lock();
try {
while (! isDone()) {
done.await(timeout, TimeUnit.MILLISECONDS);
if (isDone() || System.currentTimeMillis() - start > timeout) {
break;
}
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
} finally {
lock.unlock();
}
if (! isDone()) {
throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
}
}
return returnFromResponse();
}
public boolean isDone() {
return response != null;
}
private Object returnFromResponse() throws RemotingException {
Response res = response;
if (res == null) {
throw new IllegalStateException("response cannot be null");
}
if (res.getStatus() == Response.OK) {
return res.getResult();
}
if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
throw new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
}
throw new RemotingException(channel, res.getErrorMessage());
}
//channelHandle收到服務端響應後調用該方法喚醒調用方等待線程
public static void received(Channel channel, Response response) {
try {
DefaultFuture future = FUTURES.remove(response.getId());
if (future != null) {
future.doReceived(response);
} else {
logger.warn("The timeout response finally returned at "
+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
+ ", response " + response
+ (channel == null ? "" : ", channel: " + channel.getLocalAddress()
+ " -> " + channel.getRemoteAddress()));
}
} finally {
CHANNELS.remove(response.getId());
}
}
private void doReceived(Response res) {
lock.lock();
try {
response = res;
if (done != null) {
//發生收到相應信號，通知等待線程
done.signal();
}
} finally {
lock.unlock();
}
if (callback != null) {
invokeCallback(callback);
}
}
}
public interface ChannelHandlerDelegate extends ChannelHandler {
public ChannelHandler getHandler();
}
/* *HeaderExchangeHandler實現了ChannelHandler接口。實現了NIO消息回調處理， *本類最關鍵的方法在於received方法，同時在各個回調方法設置Chanel的讀寫數據時間用於心跳定時器任務發送心跳判斷條件 */
public class HeaderExchangeHandler implements ChannelHandlerDelegate {
protected static final Logger logger = LoggerFactory.getLogger(HeaderExchangeHandler.class);
public static String KEY_READ_TIMESTAMP = HeartbeatHandler.KEY_READ_TIMESTAMP;
public static String KEY_WRITE_TIMESTAMP = HeartbeatHandler.KEY_WRITE_TIMESTAMP;
private final ExchangeHandler handler;
public HeaderExchangeHandler(ExchangeHandler handler){
if (handler == null) {
throw new IllegalArgumentException("handler == null");
}
this.handler = handler;
}
void handlerEvent(Channel channel, Request req) throws RemotingException {
if (req.getData() != null && req.getData().equals(Request.READONLY_EVENT)) {
channel.setAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY, Boolean.TRUE);
}
}
Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
Response res = new Response(req.getId(), req.getVersion());
if (req.isBroken()) {
Object data = req.getData();
String msg;
if (data == null) msg = null;
else if (data instanceof Throwable) msg = StringUtils.toString((Throwable) data);
else msg = data.toString();
res.setErrorMessage("Fail to decode request due to: " + msg);
res.setStatus(Response.BAD_REQUEST);
return res;
}
// find handler by message class.
Object msg = req.getData();
try {
// handle data.
Object result = handler.reply(channel, msg);
res.setStatus(Response.OK);
res.setResult(result);
} catch (Throwable e) {
res.setStatus(Response.SERVICE_ERROR);
res.setErrorMessage(StringUtils.toString(e));
}
return res;
}
static void handleResponse(Channel channel, Response response) throws RemotingException {
if (response != null && !response.isHeartbeat()) {
//很是關鍵的調用，用於將client收到的響應喚醒等待響應結果的用戶線程
DefaultFuture.received(channel, response);
}
}
public void connected(Channel channel) throws RemotingException {
channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
channel.setAttribute(KEY_WRITE_TIMESTAMP, System.currentTimeMillis());
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
try {
handler.connected(exchangeChannel);
} finally {
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}
public void disconnected(Channel channel) throws RemotingException {
channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
channel.setAttribute(KEY_WRITE_TIMESTAMP, System.currentTimeMillis());
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
try {
handler.disconnected(exchangeChannel);
} finally {
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}
public void sent(Channel channel, Object message) throws RemotingException {
Throwable exception = null;
try {
channel.setAttribute(KEY_WRITE_TIMESTAMP, System.currentTimeMillis());
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
try {
handler.sent(exchangeChannel, message);
} finally {
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
} catch (Throwable t) {
exception = t;
}
if (message instanceof Request) {
Request request = (Request) message;
DefaultFuture.sent(channel, request);
}
if (exception != null) {
if (exception instanceof RuntimeException) {
throw (RuntimeException) exception;
} else if (exception instanceof RemotingException) {
throw (RemotingException) exception;
} else {
throw new RemotingException(channel.getLocalAddress(), channel.getRemoteAddress(),
exception.getMessage(), exception);
}
}
}
private static boolean isClientSide(Channel channel) {
InetSocketAddress address = channel.getRemoteAddress();
URL url = channel.getUrl();
return url.getPort() == address.getPort() &&
NetUtils.filterLocalHost(url.getIp())
.equals(NetUtils.filterLocalHost(address.getAddress().getHostAddress()));
}
//關鍵方法，實現client端Response響應對應處理以及server端Request請求處理
public void received(Channel channel, Object message) throws RemotingException {
channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
try {
if (message instanceof Request) {
// handle request.
Request request = (Request) message;
if (request.isEvent()) {
handlerEvent(channel, request);
} else {
if (request.isTwoWay()) {
Response response = handleRequest(exchangeChannel, request);
channel.send(response);
} else {
handler.received(exchangeChannel, request.getData());
}
}
} else if (message instanceof Response) {
handleResponse(channel, (Response) message);
} else if (message instanceof String) {
if (isClientSide(channel)) {
Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
logger.error(e.getMessage(), e);
} else {
String echo = handler.telnet(channel, (String) message);
if (echo != null && echo.length() > 0) {
channel.send(echo);
}
}
} else {
handler.received(exchangeChannel, message);
}
} finally {
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}
public void caught(Channel channel, Throwable exception) throws RemotingException {
if (exception instanceof ExecutionException) {
ExecutionException e = (ExecutionException) exception;
Object msg = e.getRequest();
if (msg instanceof Request) {
Request req = (Request) msg;
if (req.isTwoWay() && ! req.isHeartbeat()) {
Response res = new Response(req.getId(), req.getVersion());
res.setStatus(Response.SERVER_ERROR);
res.setErrorMessage(StringUtils.toString(e));
channel.send(res);
return;
}
}
}
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
try {
handler.caught(exchangeChannel, exception);
} finally {
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}
public ChannelHandler getHandler() {
if (handler instanceof ChannelHandlerDelegate) {
return ((ChannelHandlerDelegate) handler).getHandler();
} else {
return handler;
}
}
}
/* *千呼萬喚始出來，是時候展現dubbo協議的入口函數了 */
public class DubboProtocol extends AbstractProtocol {
//根據服務提供者URL和服務接口類型建立Invoker對象
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
// modified by lishen
optimizeSerialization(url);
// create rpc invoker.
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
invokers.add(invoker);
return invoker;
}
private ExchangeClient[] getClients(URL url){
//默認使用共享連接
boolean service_share_connect = false;
int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
//若是connections不配置，則共享鏈接，不然每服務每鏈接
if (connections == 0){
service_share_connect = true;
connections = 1;
}
ExchangeClient[] clients = new ExchangeClient[connections];
for (int i = 0; i < clients.length; i++) {
if (service_share_connect){
clients[i] = getSharedClient(url);
} else {
clients[i] = initClient(url);
}
}
return clients;
}
/** *獲取共享鏈接 */
private ExchangeClient getSharedClient(URL url){
String key = url.getAddress();
ReferenceCountExchangeClient client = referenceClientMap.get(key);
if ( client != null ){
if ( !client.isClosed()){
client.incrementAndGetCount();
return client;
} else {
// logger.warn(new IllegalStateException("client is closed,but stay in clientmap .client :"+ client));
referenceClientMap.remove(key);
}
}
ExchangeClient exchagneclient = initClient(url);
client = new ReferenceCountExchangeClient(exchagneclient, ghostClientMap);
referenceClientMap.put(key, client);
ghostClientMap.remove(key);
return client;
}
/** * 建立新鏈接. */
private ExchangeClient initClient(URL url) {
// client type setting.
String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
String version = url.getParameter(Constants.DUBBO_VERSION_KEY);
boolean compatible = (version != null && version.startsWith("1.0."));
url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() && compatible ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
//默認開啓heartbeat
url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
// BIO存在嚴重性能問題，暫時不容許使用
if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
throw new RpcException("Unsupported client type: " + str + "," +
" supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
}
ExchangeClient client ;
try {
//設置鏈接應該是lazy的
if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)){
client = new LazyConnectExchangeClient(url ,requestHandler);
} else {
//該工廠方法會建立以前介紹的HeaderExchangeClient實例
client = Exchangers.connect(url ,requestHandler);
}
} catch (RemotingException e) {
throw new RpcException("Fail to create remoting client for service(" + url
+ "): " + e.getMessage(), e);
}
return client;
}
}
/* *經過DubboProtocol建立好的ExchangeClient對象來建立DubboInvoker *DubboInvoker能夠支持異步調用，同步調用，僅發送請求3種模式 */
public class DubboInvoker<T> extends AbstractInvoker<T> {
private final ExchangeClient[] clients;
private final AtomicPositiveInteger index = new AtomicPositiveInteger();
private final String version;
private final ReentrantLock destroyLock = new ReentrantLock();
private final Set<Invoker<?>> invokers;
public DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[] clients){
this(serviceType, url, clients, null);
}
public DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[] clients, Set<Invoker<?>> invokers){
super(serviceType, url, new String[] {Constants.INTERFACE_KEY, Constants.GROUP_KEY, Constants.TOKEN_KEY, Constants.TIMEOUT_KEY});
this.clients = clients;
// get version.
this.version = url.getParameter(Constants.VERSION_KEY, "0.0.0");
this.invokers = invokers;
}
@Override
protected Result doInvoke(final Invocation invocation) throws Throwable {
RpcInvocation inv = (RpcInvocation) invocation;
final String methodName = RpcUtils.getMethodName(invocation);
inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
inv.setAttachment(Constants.VERSION_KEY, version);
ExchangeClient currentClient;
if (clients.length == 1) {
currentClient = clients[0];
} else {
currentClient = clients[index.getAndIncrement() % clients.length];
}
try {
boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT);
if (isOneway) {
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
currentClient.send(inv, isSent);
RpcContext.getContext().setFuture(null);
return new RpcResult();
} else if (isAsync) {
ResponseFuture future = currentClient.request(inv, timeout) ;
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
return new RpcResult();
} else {
RpcContext.getContext().setFuture(null);
return (Result) currentClient.request(inv, timeout).get();
}
} catch (TimeoutException e) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} catch (RemotingException e) {
throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
@Override
public boolean isAvailable() {
if (!super.isAvailable())
return false;
for (ExchangeClient client : clients){
if (client.isConnected() && !client.hasAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY)){
//cannot write == not Available ?
return true ;
}
}
return false;
}
public void destroy() {
//防止client被關閉屢次.在connect per jvm的狀況下，client.close方法會調用計數器-1，當計數器小於等於0的狀況下，才真正關閉
if (super.isDestroyed()){
return ;
} else {
//dubbo check ,避免屢次關閉
destroyLock.lock();
try{
if (super.isDestroyed()){
return ;
}
super.destroy();
if (invokers != null){
invokers.remove(this);
}
for (ExchangeClient client : clients) {
try {
client.close();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}finally {
destroyLock.unlock();
}
}
}
}
複製代碼
相關文章
- 1. 網絡通信-網絡通信過程
- 2. dubbo源碼解析（10）dubbo網絡通信
- 3. 網絡編程（四）UDP通信
- 4. 網絡通信
- 5. C#網絡通信
- 6. CAN網絡通信
- 7. 網絡通信_SSH
- 8. TCP網絡通信
- 9. 網絡通信——socket
- 10. linuxc網絡通信
- 更多相關文章...
- • netwox顯示網絡配置信息 - TCP/IP教程
- • 網絡協議是什麼？ - TCP/IP教程
- • RxJava操作符（四）Combining
- • 互聯網組織的未來：剖析GitHub員工的任性之源
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. CVPR 2020 論文大盤點-光流篇
- 2. Photoshop教程_ps中怎麼載入圖案？PS圖案如何導入？
- 3. org.pentaho.di.core.exception.KettleDatabaseException:Error occurred while trying to connect to the
- 4. SonarQube Scanner execution execution Error --- Failed to upload report - 500: An error has occurred
- 5. idea 導入源碼包
- 6. python學習 day2——基礎學習
- 7. 3D將是頁遊市場新賽道？
- 8. osg--交互
- 9. OSG-交互
- 10. Idea、spring boot 圖片(pgn顯示、jpg不顯示)解決方案
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 網絡通信-網絡通信過程
- 2. dubbo源碼解析（10）dubbo網絡通信
- 3. 網絡編程（四）UDP通信
- 4. 網絡通信
- 5. C#網絡通信
- 6. CAN網絡通信
- 7. 網絡通信_SSH
- 8. TCP網絡通信
- 9. 網絡通信——socket
- 10. linuxc網絡通信