Tear the source code of gateway by hand, and tear the source code of workflow and load balancing today

Spring Cloud Gateway Source analysis

Through the previous study , We know SpringCloud Gateway It's a microservice gateway , It mainly realizes service routing of different functions , About SpringCloud Gateway We will come to an end with the actual use of , Let's go further SpringCloud Gateway Source code .

2.1 Gateway Workflow source code analysis

2.1.1 Gateway Workflow analysis

We have learned before Gateway workflow , The work flow chart above , Let's review the workflow ：

1: All will be ReactorHttpHandlerAdapter.apply() Method intercept processing , At this time, the request object and response object will be encapsulated , And deliver it to HttpWebHandlerAdapter.handle() Method .

2:HttpWebHandlerAdapter.handle(), take request and response Encapsulated as a context object ServerWebExchange, Methods by getDelegate() Get the global exception handler ExceptionHandlingWebHandler Perform global exception handling 

3:ExceptionHandlingWebHandler After execution , call DispatcherHandler.handle(), Cycle all handlerMappings Find the Handler

4: find Handler After the call DispatcherHandler.invokeHandler() Do what you find Handler, This will call FilteringWebHandler.handle()

5:DefaultGatewayFilterChain.filter() Is the key process , All filters will be executed here , For example, service search 、 Load balancing 、 Remote call, etc , It's all here .

We are all based on the above workflow , Next, we analyze layer by layer Gateway Source code , Learn more Gateway.

2.1.2 Gateway Workflow source code

Let's take a look first Gateway Intercept the method of processing all requests handle()：

/**** * Handle all requests  ****/
@Override
public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
    
    if (this.forwardedHeaderTransformer != null) {
    
        request = this.forwardedHeaderTransformer.apply(request);
    }
    // Create a gateway context object 
    ServerWebExchange exchange = createExchange(request, response);

    LogFormatUtils.traceDebug(logger, traceOn ->
            exchange.getLogPrefix() + formatRequest(exchange.getRequest()) +
                    (traceOn ? ", headers=" + formatHeaders(exchange.getRequest().getHeaders()) : ""));

    //getDelegate() Get current Handler
    return getDelegate().handle(exchange)
            .doOnSuccess(aVoid -> logResponse(exchange))
            .onErrorResume(ex -> handleUnresolvedError(exchange, ex))
            .then(Mono.defer(response::setComplete));
}

above getDelegate() The source code of the method is as follows ：

/** * Return the wrapped delegate. *  return WebHandler： Handle web The object of the request  */
public WebHandler getDelegate() {
    
	return this.delegate;
}

We carry out Debug Test the following ：

Currently returned WebHandler yes ExceptionHandlingWebHandler, and ExceptionHandlingWebHandler Of delegate yes FilteringWebHandler, and FilteringWebHandler Of delegate yes delegate yes DispatcherHandler, be-all delegate Of handle() Methods will be executed in turn , We can put the breakpoint in DispatcherHandler.handler() On the way ：

handler() Method will call all handlerMappings Of getHandler(exchange) Method , and getHandler(exchange) Method will call getHandlerInternal(exchange) Method ：

getHandlerInternal(exchange) This method consists of various HandlerMapping To achieve , We can observe the assertion processing RoutePredicateHandlerMapping Of getHandlerInternal(exchange) Method will call lookupRoute Method , This method is used to return the corresponding routing information :

The route matching here is actually the information of a service corresponding to the route configuration in our project , These service information can help us find the real service we want to call ：

Every Route Object as follows ：

Route Of DEBUG The data are as follows ：

To find the corresponding Route The specified FilterWebHandler, The following code ：

FilterWebHandler It mainly includes all filters , Filters are sorted in a certain order , Mainly order value , The smaller, the closer to the front , In the filter, the request is mainly handled by the specified real service ,debug Test the following ：

Here you are RouteToRequestUrlFilter and ForwardRoutingFilter as well as LoadBalancerClientFilter Wait for multiple filters .

2.1.3 Request processing

on top FilterWebHandler There is 2 A filter , Respectively RouteToRequestUrlFilter and ForwardRoutingFilter .

RouteToRequestUrlFilter： For use in accordance with matching Of Route, Calculate the request address to get lb://hailtaxi-order/order/list

ForwardRoutingFilter: Forwarding route gateway filter . Its basis forward:// Prefix ( Scheme ) Filtering treatment , Forward the request to the local interface of the current gateway instance .

2.1.3.1 RouteToRequestUrlFilter Real service search

RouteToRequestUrlFilter Source code is as follows ：

/*** *  Handle uri filter  * @param exchange * @param chain * @return */
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    
    // Get current route
    Route route = exchange.getAttribute(GATEWAY_ROUTE_ATTR);
    if (route == null) {
    
        return chain.filter(exchange);
    }
    log.trace("RouteToRequestUrlFilter start");
    // obtain uri = http://localhost:8001/order/list?token=123
    URI uri = exchange.getRequest().getURI();
    boolean encoded = containsEncodedParts(uri);
    URI routeUri = route.getUri();

    if (hasAnotherScheme(routeUri)) {
    
        // this is a special url, save scheme to special attribute
        // replace routeUri with schemeSpecificPart
        exchange.getAttributes().put(GATEWAY_SCHEME_PREFIX_ATTR,
                routeUri.getScheme());
        routeUri = URI.create(routeUri.getSchemeSpecificPart());
    }

    if ("lb".equalsIgnoreCase(routeUri.getScheme()) && routeUri.getHost() == null) {
    
        // Load balanced URIs should always have a host. If the host is null it is
        // most
        // likely because the host name was invalid (for example included an
        // underscore)
        throw new IllegalStateException("Invalid host: " + routeUri.toString());
    }

    // take uri Switch to  lb://hailtaxi-order/order/list?token=123 
    URI mergedUrl = UriComponentsBuilder.fromUri(uri)
            // .uri(routeUri)
            .scheme(routeUri.getScheme()).host(routeUri.getHost())
            .port(routeUri.getPort()).build(encoded).toUri();
    exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, mergedUrl);
    return chain.filter(exchange);
}

debug The debugging results are as follows ：

From the above debugging results, we can see the selected Route as well as uri and routeUri and mergedUrl, The filter is actually to replace the address requested by the user with the service address , Changing the service address can be used for load balancing .

2.1.3.2 NettyRoutingFilter The remote invocation

SpringCloud The implementation of remote calls to back-end services is based on Netty send out Http Request to implement , The core code is NettyRoutingFilter.filter() in , The core code is send() Method , The code is as follows ：

Flux<HttpClientResponse> responseFlux = httpClientWithTimeoutFrom(route)
		//  Head information processing 
		.headers(headers -> {
    
			headers.add(httpHeaders);
			// Will either be set below, or later by Netty
			headers.remove(HttpHeaders.HOST);
			if (preserveHost) {
    
				String host = request.getHeaders().getFirst(HttpHeaders.HOST);
				headers.add(HttpHeaders.HOST, host);
			}
			//  Execute send , be based on HTTP agreement 
		}).request(method).uri(url).send((req, nettyOutbound) -> {
    
			if (log.isTraceEnabled()) {
    
				nettyOutbound
						.withConnection(connection -> log.trace("outbound route: "
								+ connection.channel().id().asShortText()
								+ ", inbound: " + exchange.getLogPrefix()));
			}
			return nettyOutbound.send(request.getBody()
					.map(dataBuffer -> ((NettyDataBuffer) dataBuffer)
							.getNativeBuffer()));
		}).
		//  In response to the results 
		responseConnection((res, connection) -> {
    

			// Defer committing the response until all route filters have run
			// Put client response as ServerWebExchange attribute and write
			// response later NettyWriteResponseFilter
			exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
			exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);
			//  Get response results 
			ServerHttpResponse response = exchange.getResponse();
			// put headers and status so filters can modify the response
			HttpHeaders headers = new HttpHeaders();

			res.responseHeaders().forEach(
					entry -> headers.add(entry.getKey(), entry.getValue()));

			String contentTypeValue = headers.getFirst(HttpHeaders.CONTENT_TYPE);
			if (StringUtils.hasLength(contentTypeValue)) {
    
				exchange.getAttributes().put(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR,
						contentTypeValue);
			}

			setResponseStatus(res, response);

			// make sure headers filters run after setting status so it is
			// available in response
			HttpHeaders filteredResponseHeaders = HttpHeadersFilter.filter(
					getHeadersFilters(), headers, exchange, Type.RESPONSE);

			if (!filteredResponseHeaders
					.containsKey(HttpHeaders.TRANSFER_ENCODING)
					&& filteredResponseHeaders
							.containsKey(HttpHeaders.CONTENT_LENGTH)) {
    
				// It is not valid to have both the transfer-encoding header and
				// the content-length header.
				// Remove the transfer-encoding header in the response if the
				// content-length header is present.
				response.getHeaders().remove(HttpHeaders.TRANSFER_ENCODING);
			}

			exchange.getAttributes().put(CLIENT_RESPONSE_HEADER_NAMES,
					filteredResponseHeaders.keySet());

			response.getHeaders().putAll(filteredResponseHeaders);

			return Mono.just(res);
		});

Duration responseTimeout = getResponseTimeout(route);

above send The method will eventually be called ChannelOperations>send() Method , This method is actually based on Netty Realize data transmission , The core code is as follows ：

2.1.3.3 Netty characteristic

Netty It's based on NIO（Nonblocking I/O, Non blocking IO） Network communication framework developed , His concurrency performance has been greatly improved , In contrast to BIO（Blocking I/O, Blocking IO）, Hide the complexity behind it and provide an easy to use API The client of / Server framework .Netty Is a widely used Java Network programming framework .

Extremely fast transmission

Netty In fact, the speed of transmission depends on NIO A feature of —— Zero copy . We know ,Java There is heap memory 、 Stack memory, string constant pool, etc , Among them, the heap memory is the largest one , It's also Java Where objects are stored , Generally, if our data needs to be collected from IO Read to heap memory , You need to go through Socket buffer , In other words, a data will be copied twice to reach its destination , If you have a lot of data , It will cause unnecessary waste of resources .
Netty In this case , Used NIO Another big feature of —— Zero copy , When he needs to receive data , He'll carve out a piece of memory outside the heap , The data comes directly from IO Read to that memory , stay netty Inside through ByteBuf You can directly manipulate these data , This speeds up the transmission .

Good packaging

Netty Both performance and encapsulation are far beyond the traditional Socket Programming .

Channel: Represents a connection , It can be understood as every request , It's just one. Channel.

ChannelHandler: The core processing business is here , Used to process business requests .

ChannelHandlerContext: Used to transmit business data .

ChannelPipeline: It is used to save the information needed in the process ChannelHandler and ChannelHandlerContext.

ByteBuf Is a container for storing bytes , The biggest feature is easy to use , It has its own read index and write index , It is convenient for you to read and write the whole byte cache , Also support get/set, It is convenient for you to read and write each byte , His data structure is shown in the figure below ：

2.2 Gateway Load balancing source code analysis

The previous source code analysis mainly analyzes Gateway workflow , Let's analyze Gateway Load balancing process . In the final filter set LoadBalancerClientFilter filter , This filter is used to realize load balancing .

2.2.1 address translation

LoadBalancerClientFilter The filter will first convert the user request address into the real service address , That is to say IP: Port number , Source code is as follows ：

/*** *  Load balancing filtering  * @param exchange * @param chain * @return */
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    
    // Load balanced URL = lb://hailtaxi-order/order/list?token=123
    URI url = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
    String schemePrefix = exchange.getAttribute(GATEWAY_SCHEME_PREFIX_ATTR);
    if (url == null
            || (!"lb".equals(url.getScheme()) && !"lb".equals(schemePrefix))) {
    
        return chain.filter(exchange);
    }
    // preserve the original url
    addOriginalRequestUrl(exchange, url);

    if (log.isTraceEnabled()) {
    
        log.trace("LoadBalancerClientFilter url before: " + url);
    }
    
    // Service selection 
    final ServiceInstance instance = choose(exchange);

    if (instance == null) {
    
        throw NotFoundException.create(properties.isUse404(),
                "Unable to find instance for " + url.getHost());
    }
    // User submitted URI = http://localhost:8001/order/list?token=123
    URI uri = exchange.getRequest().getURI();

    // if the `lb:<scheme>` mechanism was used, use `<scheme>` as the default,
    // if the loadbalancer doesn't provide one.
    String overrideScheme = instance.isSecure() ? "https" : "http";
    if (schemePrefix != null) {
    
        overrideScheme = url.getScheme();
    }
    // Real service URL =http://192.168.211.1:18182/order/list?token=123
    URI requestUrl = loadBalancer.reconstructURI(
            new DelegatingServiceInstance(instance, overrideScheme), uri);

    if (log.isTraceEnabled()) {
    
        log.trace("LoadBalancerClientFilter url chosen: " + requestUrl);
    }

    exchange.getAttributes().put(GATEWAY_REQUEST_URL_ATTR, requestUrl);
    return chain.filter(exchange);
}

2.2.2 Load balancing service selection

The key to the above code is choose(exchange) Call to , This method call is actually to select the specified service , This involves the load balancing service polling call Algorithm , We can track in to see the method execution process .

Gateway It has been integrated Ribbon, So the object we see is RibbonLoadBalancerClient, We followed in and checked ：

The above methods will be called to getInstance() Method , This method will return all available instances , There may be multiple instances , If there are multiple instances, the load balancing algorithm is involved , The method call is shown in the following figure ：

Call at this time getServer() Method , Call again BaseLoadBalancer.chooseServer(), Here is to obtain the corresponding instance according to the specified algorithm , The code is as follows ：

BaseLoadBalancer It belongs to Ribbon The algorithm of , We can know through the following dependency packages , And the default of this algorithm is RoundRobinRule, That is, random algorithm , The following code ：

focus Java Technology dry goods sharing , Welcome like-minded friends , Exchange and study together