摘要:本文主要分析使用cse提供的RestTemplate的場景,其實cse提供的rpc注解(RpcReference)的方式最後的調用邏輯和RestTemplate是殊途同歸的。
本文分享自華為雲社群《我是一個請求,我該何去何從(下)》,原文作者:向昊 。
上次我們大概了解到了服務端是怎麼處理請求的,那麼發送請求又是個什麼樣的流程了?本文主要分析使用cse提供的RestTemplate的場景,其實cse提供的rpc注解(RpcReference)的方式最後的調用邏輯和RestTemplate是殊途同歸的。
使用
使用cse提供的RestTemplate時候,是這樣初始化的:
RestTemplate restTemplate = RestTemplateBuilder.create();
restTemplate.getForObject("cse://appId:serviceName/xxx", Object.class);
我們可以注意到2個怪異的地方:
- RestTemplate是通過RestTemplateBuilder.create()來擷取的,而不是用的Spring裡提供的。
- 請求路徑開頭是cse而不是我們常見的http、https且需要加上服務所屬的應用ID和服務名稱。
解析
根據url比對RestTemplate
首先看下RestTemplateBuilder.create(),它傳回的是org.apache.servicecomb.provider.springmvc.reference.RestTemplateWrapper,是cse提供的一個包裝類。
// org.apache.servicecomb.provider.springmvc.reference.RestTemplateWrapper
// 用于同時支援cse調用和非cse調用
class RestTemplateWrapper extends RestTemplate {
private final List<AcceptableRestTemplate> acceptableRestTemplates = new ArrayList<>();
final RestTemplate defaultRestTemplate = new RestTemplate();
RestTemplateWrapper() {
acceptableRestTemplates.add(new CseRestTemplate());
}
RestTemplate getRestTemplate(String url) {
for (AcceptableRestTemplate template : acceptableRestTemplates) {
if (template.isAcceptable(url)) {
return template;
}
}
return defaultRestTemplate;
}
}
AcceptableRestTemplate:這個類是一個抽象類,也是繼承RestTemplate的,目前其子類就是CseRestTemplate,我們也可以看到在初始化的時候會預設往acceptableRestTemplates中添加一個CseRestTemplate。
回到使用的地方restTemplate.getForObject:這個方法會委托給如下方法:
public <T> T getForObject(String url, Class<T> responseType, Object... urlVariables) throws RestClientException {
return getRestTemplate(url).getForObject(url, responseType, urlVariables);
}
可以看到首先會調用getRestTemplate(url),即會調用template.isAcceptable(url),如果比對到了就傳回CseRestTemplate,否則就傳回正常的RestTemplate。那麼再看下isAcceptable()這個方法:
到這裡我們就清楚了路徑中的cse://的作用了,就是為了使用CseRestTemplate來發起請求,也了解了為啥RestTemplateWrapper可以同時支援cse調用和非cse調用。
委托調用
從上面可知,我們的cse調用其實都是委托給CseRestTemplate了。在構造CseRestTemplate的時候會初始化幾個東西:
public CseRestTemplate() {
setMessageConverters(Arrays.asList(new CseHttpMessageConverter()));
setRequestFactory(new CseClientHttpRequestFactory());
setUriTemplateHandler(new CseUriTemplateHandler());
}
這裡需要重點關注new CseClientHttpRequestFactory():
public class CseClientHttpRequestFactory implements ClientHttpRequestFactory {
@Override
public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod) throws IOException {
return new CseClientHttpRequest(uri, httpMethod);
}
}
最終委托到了CseClientHttpRequest這個類,這裡就是重頭戲了!
我們先把注意力拉回到這句話:restTemplate.getForObject("cse://appId:serviceName/xxx", Object.class),從上面我們知道其邏輯是先根據url找到對應的RestTemplate,然後調用getForObject這個方法,最終這個方法會調用到:org.springframework.web.client.RestTemplate#doExecute:
protected <T> T doExecute(URI url, @Nullable HttpMethod method, @Nullable RequestCallback requestCallback,
@Nullable ResponseExtractor<T> responseExtractor) throws RestClientException {
ClientHttpResponse response = null;
try {
ClientHttpRequest request = createRequest(url, method);
if (requestCallback != null) {
requestCallback.doWithRequest(request);
}
response = request.execute();
handleResponse(url, method, response);
return (responseExtractor != null ? responseExtractor.extractData(response) : null);
}
}
createRequest(url, method):會調用getRequestFactory().createRequest(url, method),即最終會調用到我們初始化CseClientHttpRequest是塞的RequestFactory,是以這裡會傳回ClientHttpRequest這個類。
request.execute():這個方法會委托到org.apache.servicecomb.provider.springmvc.reference.CseClientHttpRequest#execute這個方法上。
至此我們知道前面的調用最終會委托到CseClientHttpRequest#execute這個方法上。
cse調用
接着上文分析:
public ClientHttpResponse execute() {
path = findUriPath(uri);
requestMeta = createRequestMeta(method.name(), uri);
QueryStringDecoder queryStringDecoder = new QueryStringDecoder(uri.getRawSchemeSpecificPart());
queryParams = queryStringDecoder.parameters();
Object[] args = this.collectArguments();
// 異常流程,直接抛異常出去
return this.invoke(args);
}
createRequestMeta(method.name(), uri):這裡主要是根據microserviceName去擷取調用服務的資訊,并會将擷取的資訊放入到Map中。服務資訊如下:
可以看到裡面的資訊很豐富,例如應用名、服務名、還有接口對應的yaml資訊等。
this.collectArguments():這裡隐藏了一個校驗點,就是會校驗傳入的參數是否符合對方接口的定義。主要是通過這個方法:org.apache.servicecomb.common.rest.codec.RestCodec#restToArgs,如果不符合真個流程就結束了。
準備invocation
從上面分析可知,擷取到接口所需的參數後就會調用這個方法:org.apache.servicecomb.provider.springmvc.reference.CseClientHttpRequest#invoke:
private CseClientHttpResponse invoke(Object[] args) {
Invocation invocation = prepareInvocation(args);
Response response = doInvoke(invocation);
if (response.isSuccessed()) {
return new CseClientHttpResponse(response);
}
throw ExceptionFactory.convertConsumerException(response.getResult());
}
prepareInvocation(args):這個方法會準備好Invocation,這個Invocation在上集已經分析過了,不過上集中的它是為服務端服務的,那麼咱們這塊當然就得為消費端工作了
protected Invocation prepareInvocation(Object[] args) {
Invocation invocation =
InvocationFactory.forConsumer(requestMeta.getReferenceConfig(),
requestMeta.getOperationMeta(),
args);
return invocation;
}
從名字也可以看出它是為消費端服務的,其實無論是forProvider還是forConsumer,它們最主要的差別就是加載的Handler不同,這次加載的Handler如下:
- class org.apache.servicecomb.qps.ConsumerQpsFlowControlHandler(流控)
- class org.apache.servicecomb.loadbalance.LoadbalanceHandler(負載)
- class org.apache.servicecomb.bizkeeper.ConsumerBizkeeperHandler(容錯)
- class org.apache.servicecomb.core.handler.impl.TransportClientHandler(調用,預設加載的)
前面3個Handler可以參考下這個微服務治理專欄
doInvoke(invocation):初始化好了invocation後就開始調用了。最終會調用到這個方法上:org.apache.servicecomb.core.provider.consumer.InvokerUtils#innerSyncInvoke
至此,這些動作就是cse中RestTemplate和rpc調用的不同之處。不過可以清楚的看到RestTemplate的方式是隻支援同步的,即innerSyncInvoke,但是rpc是可以支援異步的,即reactiveInvoke
public static Response innerSyncInvoke(Invocation invocation) {
invocation.next(respExecutor::setResponse);
}
到這裡我們知道了,消費端發起請求還是得靠invocation的責任鍊驅動
啟動invocation責任鍊
好了,咱們的老朋友又出現了:invocation.next,這個方法是個典型的責任鍊模式,其鍊條就是上面說的那4個Handler。前面3個就不分析了,直接跳到TransportClientHandler。
// org.apache.servicecomb.core.handler.impl.TransportClientHandler
public void handle(Invocation invocation, AsyncResponse asyncResp) throws Exception {
Transport transport = invocation.getTransport();
transport.send(invocation, asyncResp);
}
invocation.getTransport():擷取請求位址,即最終發送請求的時候還是以ip:port的形式。
transport.send(invocation, asyncResp):調用鍊為
org.apache.servicecomb.transport.rest.vertx.VertxRestTransport#send
- ->org.apache.servicecomb.transport.rest.client.RestTransportClient#send(這裡會初始化HttpClientWithContext,下面會分析)
- ->org.apache.servicecomb.transport.rest.client.http.RestClientInvocation#invoke(真正發送請求的地方)
public void invoke(Invocation invocation, AsyncResponse asyncResp) throws Exception {
createRequest(ipPort, path);
clientRequest.putHeader(org.apache.servicecomb.core.Const.TARGET_MICROSERVICE, invocation.getMicroserviceName());
RestClientRequestImpl restClientRequest =
new RestClientRequestImpl(clientRequest, httpClientWithContext.context(), asyncResp, throwableHandler);
invocation.getHandlerContext().put(RestConst.INVOCATION_HANDLER_REQUESTCLIENT, restClientRequest);
Buffer requestBodyBuffer = restClientRequest.getBodyBuffer();
HttpServletRequestEx requestEx = new VertxClientRequestToHttpServletRequest(clientRequest, requestBodyBuffer);
invocation.getInvocationStageTrace().startClientFiltersRequest();
// 觸發filter.beforeSendRequest方法
for (HttpClientFilter filter : httpClientFilters) {
if (filter.enabled()) {
filter.beforeSendRequest(invocation, requestEx);
}
}
// 從業務線程轉移到網絡線程中去發送
// httpClientWithContext.runOnContext
}
createRequest(ipPort, path):根據參數初始化HttpClientRequest clientRequest,初始化的時候會傳入一個建立一個responseHandler,即對響應的處理。
注意org.apache.servicecomb.common.rest.filter.HttpClientFilter#afterReceiveResponse的調用就是在這裡埋下伏筆的,是通過回調org.apache.servicecomb.transport.rest.client.http.RestClientInvocation#processResponseBody這個方法觸發的(在建立responseHandler時候建立的)
且org.apache.servicecomb.common.rest.filter.HttpClientFilter#beforeSendRequest:這個方法的觸發我們也可以很清楚的看到在發送請求執行的。
requestEx:注意它的類型是HttpServletRequestEx,雖然名字裡面帶有Servlet,但是打開它的方法可以發現有很多我們在tomcat中那些常用的方法都直接抛出異常了,這也是一個易錯點!
httpClientWithContext.runOnContext:用來發送請求的邏輯,不過這裡還是有點繞的,下面重點分析下
httpClientWithContext.runOnContext
首先看下HttpClientWithContext的定義:
public class HttpClientWithContext {
public interface RunHandler {
void run(HttpClient httpClient);
}
private HttpClient httpClient;
private Context context;
public HttpClientWithContext(HttpClient httpClient, Context context) {
this.httpClient = httpClient;
this.context = context;
}
public void runOnContext(RunHandler handler) {
context.runOnContext((v) -> {
handler.run(httpClient);
});
}
}
從上面可知發送請求調用的是這個方法:runOnContext,參數為RunHandler接口,然後是以lambda的方式傳入的,lambda的參數為httpClient,這個httpClient又是在HttpClientWithContext的構造函數中初始化的。這個構造函數是在org.apache.servicecomb.transport.rest.client.RestTransportClient#send這個方法中初始化的(調用org.apache.servicecomb.transport.rest.client.RestTransportClient#findHttpClientPool這個方法)。
但是我們觀察調用的地方:
// 從業務線程轉移到網絡線程中去發送
httpClientWithContext.runOnContext(httpClient -> {
clientRequest.setTimeout(operationMeta.getConfig().getMsRequestTimeout());
processServiceCombHeaders(invocation, operationMeta);
try {
restClientRequest.end();
} catch (Throwable e) {
LOGGER.error(invocation.getMarker(),
"send http request failed, local:{}, remote: {}.", getLocalAddress(), ipPort, e);
fail((ConnectionBase) clientRequest.connection(), e);
}
});
其實在這塊邏輯中HttpClient是沒有被用到的,實際上發送請求的動作是restClientRequest.end()觸發的,restClientRequest是cse中的類RestClientRequestImpl,然後它包裝了HttpClientRequest(vertx中提供的),即restClientRequest.end()最終還是委托到了HttpClientRequest.end()上了。
那麼這個HttpClientRequest是怎麼被初始化的了?它是在createRequest(ipPort, path)這個方法中初始化的,即在調用org.apache.servicecomb.transport.rest.client.http.RestClientInvocation#invoke方法入口處。
初始化的邏輯如下:
clientRequest = httpClientWithContext.getHttpClient().request(method,
requestOptions, this::handleResponse)
httpClientWithContext.getHttpClient():這個方法傳回的是HttpClient,上面說的HttpClient作用就展現出來了,用來初始化了我們發送請求的關鍵先生:HttpClientRequest。那麼至此我們發送請求的整體邏輯大概就清晰了。
總結
無論是采用RestTemplate的方式還是采用rpc注解的方式來發送請求,其底層邏輯其實是一樣的。即首先根據請求資訊比對到對方的服務資訊,然後經過一些列的Handler處理,如限流、負載、容錯等等(這也是一個很好的擴充機制),最終會走到TransportClientHandler這個Handler,然後根據條件去初始化發送的request,經過HttpClientFilter的處理後就會委托給vertx的HttpClientRequest來真正的送出請求。
點選關注,第一時間了解華為雲新鮮技術~