TAF 必修课（五）：Client 端调用

private void initCommunicator(CommunicatorConfig config) throws ConfigException {
	if (inited.get()) {
		return;
	}
	lock.lock();
	try {
		if (!inited.get()) {
			try {
				ClientLogger.init(config.getLogPath(), config.getLogLevel());
				if (StringUtils.isEmpty(config.getLocator())) {
					this.id = UUID.randomUUID().toString().replaceAll("-", "");
				} else {
					this.id = UUID.nameUUIDFromBytes(config.getLocator().getBytes()).toString().replaceAll("-", "");
				}
				this.communicatorConfig = config;
				this.threadPoolExecutor = ClientPoolManager.getClientThreadPoolExecutor(config);
				inited.set(true);
			} catch (Throwable e) {
				inited.set(false);
				throw new ConfigException(e);
			}
		}
	} finally {
		lock.unlock();
	}
}

 tcp -h 127.0.0.1 -p 20000 -t 5000 -a 1 -s setDivision ---> -a 1表示active, -a 0表示nactive -s setDisivision,

protected ServantClient initClient(URL url) {
	ServantClient client = null;
	try {
		boolean tcpNoDelay = url.getParameter(Constants.TAF_CLIENT_TCPNODELAY, false);
		long connectTimeout = url.getParameter(Constants.TAF_CLIENT_CONNECTTIMEOUT, Constants.default_connect_timeout);
		long syncTimeout = url.getParameter(Constants.TAF_CLIENT_SYNCTIMEOUT, Constants.default_sync_timeout);
		long asyncTimeout = url.getParameter(Constants.TAF_CLIENT_ASYNCTIMEOUT, Constants.default_async_timeout);
		boolean udpMode = url.getParameter(Constants.TAF_CLIENT_UDPMODE, false);

		if (this.selectorManager == null) {
			this.selectorManager = ClientPoolManager.getSelectorManager(this.protocolFactory, this.threadPoolExecutor, true, udpMode, this.servantProxyConfig);
		}

		client = new ServantClient(url.getHost(), url.getPort(), this.selectorManager, udpMode);
		client.setConnectTimeout(connectTimeout);
		client.setSyncTimeout(syncTimeout);
		client.setAsyncTimeout(asyncTimeout);
		client.setTcpNoDelay(tcpNoDelay);

		//将真正向服务端发起连接延迟到调用之前
	} catch (Throwable e) {
		throw new ProxyException(servantProxyConfig.getSimpleObjectName(), "Fail to create taf client|" + url.toIdentityString() + "|" + e.getLocalizedMessage(), e);
	}
	return client;
}

private ServantClient getJceClient() {
	return clients.length == 1 ? clients[0] : clients[(index.getAndIncrement() & Integer.MAX_VALUE) % clients.length];
}

private <T> Object createProxy(Class<T> clazz, ObjectProxy<T> objectProxy) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), 
		new Class[] { clazz, ServantProxy.class }, objectProxy);
}

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
	String methodName = method.getName();
	Class<?>[] parameterTypes = method.getParameterTypes();
	//构造一次RPC调用的上下文
	Invocation inv = protocolInvoker.invocation(proxy, method, args);
	try {
		 //负载均衡容错，从服务列表中选择一台
		Invoker<T> invoker = loadBalancer.select(protocolInvoker.getInvokers(), inv);
		return invoker.invoke(inv);
	} catch (Throwable e) {
		if (ClientLogger.getLogger().isDebugEnabled()) {
			ClientLogger.getLogger().debug(servantProxyConfig.getSimpleObjectName() + " error occurred on invoke|" + e.getLocalizedMessage(), e);
		}
		if (e instanceof NoInvokerException) {
			throw new NoConnectionException(servantProxyConfig.getSimpleObjectName(), e.getLocalizedMessage(), e);
		}
		throw new ProxyException(servantProxyConfig.getSimpleObjectName(), e.getLocalizedMessage(), e);
	}
}

protected Object doInvokeServant(final ServantInvocation inv) throws Throwable {
	long begin = System.currentTimeMillis();

	int ret = Constants.INVOKE_STATUS_SUCC;
	boolean isAsync = JceHelper.isAsync(inv.getMethodName());
	boolean isPromiseAsync = JceHelper.isPromiseAsync(inv.getMethodName());
	boolean isOneWay = false;
	String simpleMethodName = inv.getMethodName().replaceAll("promise_async_|async_", "");

	try {
		Method method = getApi().getMethod(inv.getMethodName(), inv.getParameterTypes());
		//callback异步调用
		if (isAsync) {
			isOneWay = invokeWithAsync(method, inv.getArguments(), inv.getAttachments());
			return null;
		}
		//Promise/Future调用
		else if (isPromiseAsync) {
			return invokeWithFuture(method, inv.getArguments(), inv.getAttachments());
		}
		//同步调用
		else {
			JceServiceResponse response = invokeWithSync(method, inv.getArguments(), inv.getAttachments());
			ret = response.getRet() == JceHelper.JCESERVERSUCCESS ? Constants.INVOKE_STATUS_SUCC : Constants.INVOKE_STATUS_EXEC;
			if (response.getRet() != JceHelper.JCESERVERSUCCESS) {
				throw ServerException.makeException(response.getRet());
			}
			return response.getResult();
		}
	} catch (Throwable e) {
		if (e instanceof TimeoutException) {
			ret = Constants.INVOKE_STATUS_TIMEOUT;
		} else if (e instanceof NotConnectedException) {
			ret = Constants.INVOKE_STATUS_NETCONNECTTIMEOUT;
		} else {
			ret = Constants.INVOKE_STATUS_EXEC;
		}
		throw e;
	} finally {
		if (!(isAsync || isPromiseAsync) || ret != Constants.INVOKE_STATUS_SUCC) {
			/** 同步调用或者发起异步调用失败 */
			/** 死活检查 */
			setAvailable(ServantnvokerAliveChecker.isAlive(getUrl(), config, ret));
			/** 上报proxy的调用情况*/
			InvokeStatHelper.getInstance().addProxyStat(objName).addInvokeTime(config.getModuleName(), objName, config.getSetDivision(), simpleMethodName, getUrl().getHost(), getUrl().getPort(), ret, System.currentTimeMillis() - begin);
		} else if (isOneWay) {
			/** 发起异步调用成功，但是无callback */
			/** 死活检查 */
			setAvailable(ServantnvokerAliveChecker.isAlive(getUrl(), config, ret));
		}
		//其余的异步调用在callback中进行死活检查和上报proxy调用情况
	}
}

public <T extends ServantResponse> T invokeWithSync(ServantRequest request) throws IOException {
	Ticket<T> ticket = null;
	T response = null;
	try {
		ensureConnected();
		request.setInvokeStatus(InvokeStatus.SYNC_CALL);
		ticket = TicketManager.createTicket(request, session, this.syncTimeout);

		Session current = session;
		current.write(request);
		//等待直到服务端响应回包或调用超时
		if (!ticket.await(this.syncTimeout, TimeUnit.MILLISECONDS)) {
			if (current != null && current.getStatus() != SessionStatus.CLIENT_CONNECTED) {
				throw new IOException("Connection reset by peer|" + this.getAddress());
			} else {
				throw new TimeoutException("the operation has timeout, " + this.syncTimeout + "ms|" + this.getAddress());
			}
		}
		response = ticket.response();
		if (response == null) {
			throw new IOException("the operation is failed.");
		}
		return response;
	} catch (InterruptedException e) {
		ClientLogger.getLogger().error(e.getLocalizedMessage());
	} finally {
		if (ticket != null) {
			TicketManager.removeTicket(ticket.getTicketNumber());
		}
	}
	return response;
}

public <T extends ServantResponse> void invokeWithAsync(ServantRequest request, Callback<T> callback) throws IOException {
	Ticket<T> ticket = null;

	try {
		ensureConnected();
		request.setInvokeStatus(InvokeStatus.ASYNC_CALL);
		ticket = TicketManager.createTicket(request, session, this.asyncTimeout, callback);

		Session current = session;
		current.write(request);
		//不等待
	} catch (Exception ex) {
		if (ticket != null) {
			TicketManager.removeTicket(ticket.getTicketNumber());
		}
		throw new IOException("error occurred on invoker with async", ex);
	}
}
//接收到响应回包
public void notifyResponse(T response)
{
	this.response = response;
	if (this.callback != null) this.callback.onCompleted(response);
	if (ticketListener != null) ticketListener.onResponseReceived(this);
}
//TicketManager定期检查回收
static 
{
	Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(new Runnable()
	{
		long currentTime = -1;
		
		public void run() 
		{
			Collection<Ticket<?>> values = tickets.values();
			currentTime = System.currentTimeMillis();
			for(Ticket<?> t : values)
			{
				if ((currentTime - t.startTime) > t.timeout) {
					removeTicket(t.getTicketNumber());
					t.expired();
				}
			}
		}
		
	}, 500, 500, TimeUnit.MILLISECONDS);
}
//过期处理
public void expired()
{
	this.expired = true;
	if (callback != null) callback.onExpired();
	this.countDown();
	if (ticketListener != null) ticketListener.onResponseExpired(this);
}

public <T extends ServantResponse, V> Future<V> invokeWithFuture(ServantRequest request,
																 Callback<T> callback) throws IOException {
	Ticket<T> ticket = null;
	try {
		ensureConnected();
		request.setInvokeStatus(InvokeStatus.FUTURE_CALL);
		ticket = TicketManager.createTicket(request, session, this.syncTimeout, callback);

		Session current = session;
		current.write(request);
		return new JcePromiseFuture<T, V>(ticket);
	} catch (Exception ex) {
		if (ticket != null) {
			TicketManager.removeTicket(ticket.getTicketNumber());
		}
		throw new IOException("error occurred on invoker with future", ex);
	}
}