异步调用

@param request       请求对象
@param rpcContext    RPC内置上下文
@param invokeContext 调用上下文
@param timeoutMillis 超时时间（毫秒）
@throws RemotingException    远程调用异常
@throws InterruptedException 中断异常
@since 5.2.0
510310
protected ResponseFuture doInvokeAsync(SofaRequest request, RpcInternalContext rpcContext,
                                           InvokeContext invokeContext, int timeoutMillis)
        throws RemotingException, InterruptedException {
        SofaResponseCallback listener = request.getSofaResponseCallback();
        if (listener != null) {
            // callback调用
            InvokeCallback callback = new BoltInvokerCallback(transportConfig.getConsumerConfig(),
                transportConfig.getProviderInfo(), listener, request, rpcContext,
                ClassLoaderUtils.getCurrentClassLoader());
            // 发起调用
            RPC_CLIENT.invokeWithCallback(url, request, invokeContext, callback, timeoutMillis);
            return null;
        } else {
            // future 转为 callback
            BoltResponseFuture future = new BoltResponseFuture(request, timeoutMillis);
            InvokeCallback callback = new BoltFutureInvokeCallback(transportConfig.getConsumerConfig(),
                transportConfig.getProviderInfo(), future, request, rpcContext,
                ClassLoaderUtils.getCurrentClassLoader());
            // 发起调用
            RPC_CLIENT.invokeWithCallback(url, request, invokeContext, callback, timeoutMillis);
            future.setSentTime();
            return future;
        }
    }
724024
@Override
    public <R, T extends Object> R callMethod(final String methodName, final T argument, final long timeout) throws CouldNotPerformException, InterruptedException {

        final String shortArgument = RPCHelper.argumentToString(argument);
        validateMiddleware();
        long retryTimeout = METHOD_CALL_START_TIMEOUT;
        long validTimeout = timeout;

        try {
            logger.debug("Calling method [" + methodName + "(" + shortArgument + ")] on scope: " + remoteServer.getScope().toString());
            if (!isConnected()) {
                waitForConnectionState(CONNECTED, timeout);
            }

            if (timeout > -1) {
                retryTimeout = Math.min(METHOD_CALL_START_TIMEOUT, validTimeout);
            }

            while (true) {

                if (!isActive()) {
                    throw new InvalidStateException("Remote service is not active!");
                }

                try {
                    logger.debug("Calling method [" + methodName + "(" + shortArgument + ")] on scope: " + remoteServer.getScope().toString());
                    remoteServerWatchDog.waitForServiceActivation(timeout, TimeUnit.MILLISECONDS);
                    final R returnValue = remoteServer.call(methodName, argument, retryTimeout);

                    if (retryTimeout != METHOD_CALL_START_TIMEOUT && retryTimeout > 15000) {
                        logger.info("Method[" + methodName + "(" + shortArgument + ")] returned! Continue processing...");
                    }
                    return returnValue;

                } catch (TimeoutException ex) {

                    // check if timeout is set and handle
                    if (timeout != -1) {
                        validTimeout -= retryTimeout;
                        if (validTimeout <= 0) {
                            ExceptionPrinter.printHistory(ex, logger, LogLevel.WARN);
                            throw new TimeoutException("Could not call remote Method[" + methodName + "(" + shortArgument + ")] on Scope[" + remoteServer.getScope() + "] in Time[" + timeout + "ms].");
                        }
                        retryTimeout = Math.min(generateTimeout(retryTimeout), validTimeout);
                    } else {
                        retryTimeout = generateTimeout(retryTimeout);
                    }

                    // only print warning if timeout is too long.
                    if (retryTimeout > 15000) {
                        ExceptionPrinter.printHistory(ex, logger, LogLevel.WARN);
                        logger.warn("Waiting for RPCServer[" + remoteServer.getScope() + "] to call method [" + methodName + "(" + shortArgument + ")]. Next retry timeout in " + (int) (Math.floor(retryTimeout / 1000)) + " sec.");
                    } else {
                        ExceptionPrinter.printHistory(ex, logger, LogLevel.DEBUG);
                        logger.debug("Waiting for RPCServer[" + remoteServer.getScope() + "] to call method [" + methodName + "(" + shortArgument + ")]. Next retry timeout in " + (int) (Math.floor(retryTimeout / 1000)) + " sec.");
                    }

                    Thread.yield();
                }
            }
        } catch (TimeoutException ex) {
            throw ex;
        } catch (CouldNotPerformException ex) {
            throw new CouldNotPerformException("Could not call remote Method[" + methodName + "(" + shortArgument + ")] on Scope[" + remoteServer.getScope() + "].", ex);
        }
    }
194619
private void recordDeserializeResponse(RpcResponseCommand responseCommand, InvokeContext invokeContext) {
        if (!RpcInternalContext.isAttachmentEnable()) {
            return;
        }
        RpcInternalContext context = null;
        if (invokeContext != null) {
            // 客户端异步调用的情况下，上下文会放在InvokeContext中传递
            context = invokeContext.get(RemotingConstants.INVOKE_CTX_RPC_CTX);
        }
        if (context == null) {
            context = RpcInternalContext.getContext();
        }
        int cost = context.getStopWatch().tick().read();
        int respSize = RpcProtocol.getResponseHeaderLength()
            + responseCommand.getClazzLength()
            + responseCommand.getContentLength()
            + responseCommand.getHeaderLength();
        // 记录响应反序列化大小和响应反序列化耗时
        context.setAttachment(RpcConstants.INTERNAL_KEY_RESP_SIZE, respSize);
        context.setAttachment(RpcConstants.INTERNAL_KEY_RESP_DESERIALIZE_TIME, cost);
    }
420139
public synchronized Response<ByteArray, Object> responseQueuePoll(long timeout,
                                                                      TimeUnit timeUnit)
            throws InterruptedException {
        long timeoutMs = timeUnit.toMillis(timeout);
        long timeoutWallClockMs = System.currentTimeMillis() + timeoutMs;
        while(responseQueue.isEmpty() && System.currentTimeMillis() < timeoutWallClockMs) {
            long remainingMs = Math.max(0, timeoutWallClockMs - System.currentTimeMillis());
            if(logger.isDebugEnabled()) {
                logger.debug("Start waiting for response queue with timeoutMs: " + timeoutMs);
            }
            this.wait(remainingMs);
            if(logger.isDebugEnabled()) {
                logger.debug("End waiting for response queue with timeoutMs: " + timeoutMs);
            }
        }
        return responseQueue.poll();
    }
937342
public static void carryWithRequest(RpcInvokeContext context, SofaRequest request) {
        if (context != null) {
            Map<String, String> requestBaggage = context.getAllRequestBaggage();
            if (CommonUtils.isNotEmpty(requestBaggage)) { // 需要透传
                request.addRequestProp(RemotingConstants.RPC_REQUEST_BAGGAGE, requestBaggage);
            }
        }
    }
12358
private void queueRequest(ManagedConnectionFactory managedConnectionFactory, long waitTimeout) throws ResourceAllocationException, ConnectionWaitTimeoutException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(this, tc, "queueRequest", waitTimeout);
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            Tr.debug(tc, "Current connection pool", pm.toString());
        if (waitTimeout == 0) {
            --pm.waiterCount;
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(
                         tc,
                         "Timeout.  Decremented waiterCount, which is now "
                             + pm.waiterCount
                             + " on datasource "
                             + gConfigProps.cfName);
            }
            Object[] connWaitTimeoutparms = new Object[] { "queueRequest", gConfigProps.cfName, 0, pm.waiterCount,
                                                           pm.totalConnectionCount.get() };
            Tr.error(
                     tc,
                     "POOL_MANAGER_EXCP_CCF2_0001_J2CA0045",
                     connWaitTimeoutparms);
            String connWaitTimeoutMessage = Tr.formatMessage(tc, "POOL_MANAGER_EXCP_CCF2_0001_J2CA0045", connWaitTimeoutparms);
            ConnectionWaitTimeoutException cwte = new ConnectionWaitTimeoutException(connWaitTimeoutMessage);
            com.ibm.ws.ffdc.FFDCFilter.processException(cwte, J2CConstants.DMSID_MAX_CONNECTIONS_REACHED, "192", this.pm);
            pm.activeRequest.decrementAndGet();

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                Tr.exit(this, tc, "queueRequest", cwte);
            throw cwte;
        }
        if ((tc.isDebugEnabled())) {
            ++freePoolQueuedRequests;
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                String poolStats = pm.gatherPoolStatisticalData();
                Tr.debug(this, tc, "Queueing Waiter for pool <" + gConfigProps.getXpathId() + ">. Current Pool Stats are:");
                Tr.debug(this, tc, poolStats);
            }
        }
        try {
            if (pm.displayInfiniteWaitMessage) {
                Tr.info(tc, "INFINITE_CONNECTION_WAIT_TIMEOUT_J2CA0127", gConfigProps.getXpathId());

                pm.displayInfiniteWaitMessage = false; // only display this message once per PM.
            }
            pm.activeRequest.decrementAndGet();
            if (waitTimeout < 0) {
                pm.waiterFreePoolLock.wait(0); //wait an infinite amount of time
            } else {
                pm.waiterFreePoolLock.wait(waitTimeout); //wait the specified amount of time
            }
            pm.requestingAccessToPool();
        } catch (InterruptedException ie) {
            if (tc.isDebugEnabled()) {
                Tr.debug(this, tc, "Interupted waiting for a connection");
            }

            synchronized (pm.waiterFreePoolLock) {
                --pm.waiterCount;
            }

            if (tc.isDebugEnabled()) {
                if (pm.waiterCount == 0) {
                    pm.waitersEndedTime = System.currentTimeMillis();
                    Tr.debug(this, tc, "Waiters: requests for connections are no longer being queued. End Time:" + pm.waitersEndedTime);
                    Tr.debug(this, tc, "Waiters: total time waiter were in queue: " + (pm.waitersEndedTime - pm.waitersStartedTime));
                }
            }
            ResourceAllocationException throwMe = new ResourceAllocationException(ie.getMessage());
            throwMe.initCause(ie);
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                Tr.exit(this, tc, "queueRequest", throwMe);
            throw throwMe;
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(this, tc, "queueRequest");
        }
    }
1052870
private static RequestConfig getRequestConfig(Integer readTimeoutInMillis) {
        return RequestConfig.custom().setConnectionRequestTimeout(600)
                .setConnectTimeout(XianConfig.getIntValue("apache.httpclient.connectTimeout", 600))
                .setSocketTimeout(readTimeoutInMillis).build();
    }
80332
@Override
    public V poll(long timeout, TimeUnit unit) throws InterruptedException {
        return get(pollAsync(timeout, unit));
    }
1359958
public String invoke(String message, int timeoutSeconds)
    throws ProcessException {
        try {
            HttpHelper httpHelper = new HttpHelper(getURL(destination));
            httpHelper.getConnection().setReadTimeout(timeoutSeconds * 1000);
            httpHelper.getConnection().setConnectTimeout(timeoutSeconds * 1000);
            String res = httpHelper.post(message);
            return res==null?null:res.trim();
        } catch (Exception e) {
            throw new ProcessException(-1, e.getMessage(), e);
        }
    }
529415
private void invoke(final Request<?> marshalledRequest, final Bootstrap b, final PutMediaResponseHandler responseHandler) throws InterruptedException {
        b.connect().addListener(new GenericFutureListener<ChannelFuture>() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                try {
                    Channel channel = future.channel();

                    HttpRequest request = RequestAdapter.adapt(marshalledRequest);

                    // Send request
                    channel.writeAndFlush(request);

                    // Send chunked content
                    channel.writeAndFlush(new HttpChunkedInput(new ChunkedStream(marshalledRequest.getContent())));
                } catch (Exception e) {
                    responseHandler.onFailure(e);
                }
            }
        });
    }
665353
public <T extends Response> T sendrecv ( Request request, T response, Set<RequestParam> params ) throws IOException {
        if ( isDisconnected() && this.state != 5 ) {
            throw new TransportException("Transport is disconnected " + this.name);
        }
        try {
            long timeout = !params.contains(RequestParam.NO_TIMEOUT) ? getResponseTimeout(request) : 0;

            long firstKey = doSend(request, response, params, timeout);

            if ( Thread.currentThread() == this.thread ) {
                // we are in the transport thread, ie. on idle disconnecting
                // this is synchronous operation
                // This does not handle compound requests
                synchronized ( this.inLock ) {
                    Long peekKey = peekKey();
                    if ( peekKey == firstKey ) {
                        doRecv(response);
                        response.received();
                        return response;
                    }
                    doSkip(peekKey);
                }
            }

            return waitForResponses(request, response, timeout);
        }
        catch ( IOException ioe ) {
            log.warn("sendrecv failed", ioe);
            try {
                disconnect(true);
            }
            catch ( IOException ioe2 ) {
                ioe.addSuppressed(ioe2);
                log.info("disconnect failed", ioe2);
            }
            throw ioe;
        }
        catch ( InterruptedException ie ) {
            throw new TransportException(ie);
        }
        finally {
            Response curResp = response;
            Request curReq = request;
            while ( curResp != null ) {
                this.response_map.remove(curResp.getMid());
                Request next = curReq.getNext();
                if ( next != null ) {
                    curReq = next;
                    curResp = next.getResponse();
                }
                else {
                    break;
                }
            }
        }
    }
469739
public SwitchYardServiceResponse invoke(SwitchYardServiceRequest request) {
        Map<String, Object> contextOut = new HashMap<String, Object>();
        Object contentOut = null;
        Object fault = null;
        try {
            QName serviceName = request.getServiceName();
            if (serviceName == null) {
                throw CommonKnowledgeMessages.MESSAGES.serviceNameNull();
            } else if (Strings.trimToNull(serviceName.getNamespaceURI()) == null) {
                String tns = getTargetNamespace();
                if (tns != null) {
                    serviceName = XMLHelper.createQName(tns, serviceName.getLocalPart());
                }
            }
            ServiceDomain serviceDomain = getServiceDomain();
            if (serviceDomain == null) {
                throw CommonKnowledgeMessages.MESSAGES.serviceDomainNull();
            }
            ServiceReference serviceReference = serviceDomain.getServiceReference(serviceName);
            if (serviceReference == null) {
                throw CommonKnowledgeMessages.MESSAGES.serviceReferenceNull(serviceName.toString());
            }
            final Exchange exchangeIn;
            FaultHandler handler = new FaultHandler();
            String operationName = request.getOperationName();
            if (operationName != null) {
                exchangeIn = serviceReference.createExchange(operationName, handler);
            } else {
                exchangeIn = serviceReference.createExchange(handler);
            }
            Message messageIn = exchangeIn.createMessage();
            Context contextIn = exchangeIn.getContext(messageIn);
            for (Map.Entry<String, Object> entry : request.getContext().entrySet()) {
                contextIn.setProperty(entry.getKey(), entry.getValue());
            }
            Object contentIn = request.getContent();
            if (contentIn != null) {
                messageIn.setContent(contentIn);
            }
            exchangeIn.send(messageIn);
            if (ExchangePattern.IN_OUT.equals(exchangeIn.getContract().getConsumerOperation().getExchangePattern())) {
                Exchange exchangeOut = handler.waitForOut();
                Message messageOut = exchangeOut.getMessage();
                contentOut = messageOut.getContent();
                for (Property property : exchangeOut.getContext(messageOut).getProperties()) {
                    contextOut.put(property.getName(), property.getValue());
                }
            }
            fault = handler.getFault();
        } catch (Throwable t) {
            fault = t;
        }
        return new SwitchYardServiceResponse(contentOut, contextOut, fault);
    }
1163747
public Invocation getInvocation(final RpcRequest request, InterfaceDescriptor<?> descriptor) {
		if (request == null) throw new NullPointerException("request");
		if (descriptor == null) throw new NullPointerException("descriptor");

		Invocation invocation = null;
		LinkedList<String> parts = splitPath(request.getRelativePath());

		while (!parts.isEmpty()) {
			String part = parts.removeFirst();

			// Find a method by a name.
			MethodDescriptor<?, ?> method = descriptor.getMethod(part);
			if (method == null) {
				throw RpcException.badRequest("Method is not found: " + part);
			}

			// Check the required HTTP method.
			if (method.isPost() && !request.isPost()) {
				throw RpcException.methodNotAllowed("Method not allowed, POST required");
			}

			// Parse arguments.
			List<Object> args = readArgs(method, parts, request.getQuery(), request.getPost());

			// Create a root invocation,
			// or a next invocation in a chain.
			invocation = invocation != null ? invocation.next(method, args.toArray())
			                                : Invocation.root(method, args.toArray());
			if (method.isTerminal()) {
				break;
			}

			// It's an interface method.
			// Get the next interface and proceed parsing the parts.
			descriptor = (InterfaceDescriptor<?>) method.getResult();
		}

		if (!parts.isEmpty()) {
			// No more interface descriptors in a chain, but the parts are still present.
			throw RpcException.badRequest("Failed to parse an invocation chain");
		}

		if (invocation == null) {
			throw RpcException.badRequest("Invocation chain required");
		}

		if (!invocation.getMethod().isTerminal()) {
			throw RpcException.badRequest("The last method must be a terminal one. "
					+ "It must return a data type or be void.");
		}

		return invocation;
	}
1008150
protected void beforeSend(RpcInternalContext context, SofaRequest request) {
        currentRequests.incrementAndGet();
        context.getStopWatch().tick().read();
        context.setLocalAddress(localAddress());
        if (EventBus.isEnable(ClientBeforeSendEvent.class)) {
            EventBus.post(new ClientBeforeSendEvent(request));
        }
    }
137216
private static RLSSuspendToken suspendRLS(int timeout) throws RLSTimeoutRangeException
    {
        if (tc.isEntryEnabled()) Tr.entry(tc, "suspendRLS", new Integer(timeout));
        
        if (timeout < 0 || timeout > 1000000000)
        {
             if (tc.isEventEnabled()) Tr.event(tc, "Timeout value is out of range - throwing RLSTimeoutRangeException", new Integer(timeout));
             if (tc.isEntryEnabled()) Tr.exit(tc, "suspend", "RLSTimeoutRangeException");
        	throw new RLSTimeoutRangeException();
        }
        
        RLSSuspendToken token = null;
        
        synchronized(SUSPEND_LOCK)
        {
        	if(!Configuration._isSnapshotSafe)
        	{
        		// If the server is not configured to be snapshot safe
        		// then we should put out a warning message
        		Tr.warning(tc, "CWRLS0020_SNAPSHOT_SAFE");
        	}
        	
            // Notify the token manager that the RLS has been called to suspend
            token = _tokenManager.registerSuspend(timeout);
            
            Tr.info(tc, "CWRLS0001_SUSPEND_RLS", token);
                
            // RLS has not already been suspended
            // Set suspended flag true
            _isSuspended = true;
        }
        
        if (tc.isEntryEnabled()) Tr.exit(tc, "suspendRLS", token);
        
        // Return the RLSSuspendToken
        return token;
    }
1043392
@Override
    public IPromise timeoutIn(long millis) {
        final Actor actor = Actor.sender.get();
        if ( actor != null )
            actor.delayed(millis, ()-> timedOut(Timeout.INSTANCE));
        else {
            Actors.delayedCalls.schedule( new TimerTask() {
                @Override
                public void run() {
                    timedOut(Timeout.INSTANCE);
                }
            },millis);
        }
        return this;
    }
499291
RpcRequest getRpcRequest(final HttpServletRequest request) {
		String method = request.getMethod();
		String relativePath = getRelativePath(request);

		// In servlets we cannot distinguish between query and post params,
		// so we use the same map for both. It is safe because Pdef HTTP RPC
		// always uses only one of them.
		Map<String, String> params = getParams(request);

		return new RpcRequest()
				.setMethod(method)
				.setRelativePath(relativePath)
				.setQuery(params)
				.setPost(params);
	}