zoukankan      html  css  js  c++  java
  • RestTemplate使用不当引发的问题分析

    背景

    • 系统: SpringBoot开发的Web应用;
    • ORM: JPA(Hibernate)
    • 接口功能简述: 根据实体类ID到数据库中查询实体信息,然后使用RestTemplate调用外部系统接口获取数据。

    问题现象

    1. 浏览器页面有时报504 GateWay Timeout错误,刷新多次后,则总是timeout
    2. 数据库连接池报连接耗尽异常
    3. 调用外部系统时有时报502 Bad GateWay错误

    分析过程

    为便于描述将本系统称为A,外部系统称为B。

    这三个问题环环相扣,导火索是第3个问题,然后导致第2个问题,最后导致出现第3个问题;
    原因简述: 第3个问题是由于Nginx负载下没有挂系统B,导致本系统在请求外部系统时报502错误,而A没有正确处理异常,导致http请求无法正常关闭,而springboot默认打开openSessionInView, 只有调用A的请求关闭时才会关闭数据库连接,而此时调用A的请求没有关闭,导致数据库连接没有关闭。

    这里主要分析第1个问题:为什么请求A的连接出现504 Timeout.

    AbstractConnPool

    通过日志看到A在调用B时出现阻塞,直到timeout,打印出线程堆栈查看:

    线程阻塞在AbstractConnPool类getPoolEntryBlocking方法中

        private E getPoolEntryBlocking(
                final T route, final Object state,
                final long timeout, final TimeUnit timeUnit,
                final Future<E> future) throws IOException, InterruptedException, TimeoutException {
    
            Date deadline = null;
            if (timeout > 0) {
                deadline = new Date (System.currentTimeMillis() + timeUnit.toMillis(timeout));
            }
            this.lock.lock();
            try {
               //根据route获取route对应的连接池
                final RouteSpecificPool<T, C, E> pool = getPool(route);
                E entry;
                for (;;) {
                    Asserts.check(!this.isShutDown, "Connection pool shut down");
                    for (;;) {
                       //获取可用的连接
                        entry = pool.getFree(state);
                        if (entry == null) {
                            break;
                        }
                        // 判断连接是否过期,如过期则关闭并从可用连接集合中删除
                        if (entry.isExpired(System.currentTimeMillis())) {
                            entry.close();
                        }
                        if (entry.isClosed()) {
                            this.available.remove(entry);
                            pool.free(entry, false);
                        } else {
                            break;
                        }
                    }
                   // 如果从连接池中获取到可用连接,更新可用连接和待释放连接集合
                    if (entry != null) {
                        this.available.remove(entry);
                        this.leased.add(entry);
                        onReuse(entry);
                        return entry;
                    }
    
                    // 如果没有可用连接,则创建新连接
                    final int maxPerRoute = getMax(route);
                    // 创建新连接之前,检查是否超过每个route连接池大小,如果超过,则删除可用连接集合相应数量的连接(从总的可用连接集合和每个route的可用连接集合中删除)
                    final int excess = Math.max(0, pool.getAllocatedCount() + 1 - maxPerRoute);
                    if (excess > 0) {
                        for (int i = 0; i < excess; i++) {
                            final E lastUsed = pool.getLastUsed();
                            if (lastUsed == null) {
                                break;
                            }
                            lastUsed.close();
                            this.available.remove(lastUsed);
                            pool.remove(lastUsed);
                        }
                    }
    
                    if (pool.getAllocatedCount() < maxPerRoute) {
                       //比较总的可用连接数量与总的可用连接集合大小,释放多余的连接资源
                        final int totalUsed = this.leased.size();
                        final int freeCapacity = Math.max(this.maxTotal - totalUsed, 0);
                        if (freeCapacity > 0) {
                            final int totalAvailable = this.available.size();
                            if (totalAvailable > freeCapacity - 1) {
                                if (!this.available.isEmpty()) {
                                    final E lastUsed = this.available.removeLast();
                                    lastUsed.close();
                                    final RouteSpecificPool<T, C, E> otherpool = getPool(lastUsed.getRoute());
                                    otherpool.remove(lastUsed);
                                }
                            }
                           // 真正创建连接的地方
                            final C conn = this.connFactory.create(route);
                            entry = pool.add(conn);
                            this.leased.add(entry);
                            return entry;
                        }
                    }
    
                   //如果已经超过了每个route的连接池大小,则加入队列等待有可用连接时被唤醒或直到某个终止时间
                    boolean success = false;
                    try {
                        if (future.isCancelled()) {
                            throw new InterruptedException("Operation interrupted");
                        }
                        pool.queue(future);
                        this.pending.add(future);
                        if (deadline != null) {
                            success = this.condition.awaitUntil(deadline);
                        } else {
                            this.condition.await();
                            success = true;
                        }
                        if (future.isCancelled()) {
                            throw new InterruptedException("Operation interrupted");
                        }
                    } finally {
                        //如果到了终止时间或有被唤醒时,则出队,加入下次循环
                        pool.unqueue(future);
                        this.pending.remove(future);
                    }
                    // 处理异常唤醒和超时情况
                    if (!success && (deadline != null && deadline.getTime() <= System.currentTimeMillis())) {
                        break;
                    }
                }
                throw new TimeoutException("Timeout waiting for connection");
            } finally {
                this.lock.unlock();
            }
        }
    

    getPoolEntryBlocking方法用于获取连接,主要有三步:(1).检查可用连接集合中是否有可重复使用的连接,如果有则获取连接,返回. (2)创建新连接,注意同时需要检查可用连接集合(分为每个route的和全局的)是否有多余的连接资源,如果有,则需要释放。(3)加入队列等待;

    从线程堆栈可以看出,第1个问题是由于走到了第3步。开始时是有时会报504异常,刷新多次后会一直报504异常,经过跟踪调试发现前几次会成功获取到连接,而连接池满后,后面的请求会阻塞。正常情况下当前面的连接释放到连接池后,后面的请求会得到连接资源继续执行,可现实是后面的连接一直处于等待状态,猜想可能是由于连接一直未释放导致。

    我们来看一下连接在什么时候会释放。

    RestTemplate

    由于在调外部系统B时,使用的是RestTemplate的getForObject方法,从此入手跟踪调试看一看。

    	@Override
    	public <T> T getForObject(String url, Class<T> responseType, Object... uriVariables) throws RestClientException {
    		RequestCallback requestCallback = acceptHeaderRequestCallback(responseType);
    		HttpMessageConverterExtractor<T> responseExtractor =
    				new HttpMessageConverterExtractor<T>(responseType, getMessageConverters(), logger);
    		return execute(url, HttpMethod.GET, requestCallback, responseExtractor, uriVariables);
    	}
    
    	@Override
    	public <T> T getForObject(String url, Class<T> responseType, Map<String, ?> uriVariables) throws RestClientException {
    		RequestCallback requestCallback = acceptHeaderRequestCallback(responseType);
    		HttpMessageConverterExtractor<T> responseExtractor =
    				new HttpMessageConverterExtractor<T>(responseType, getMessageConverters(), logger);
    		return execute(url, HttpMethod.GET, requestCallback, responseExtractor, uriVariables);
    	}
    
    	@Override
    	public <T> T getForObject(URI url, Class<T> responseType) throws RestClientException {
    		RequestCallback requestCallback = acceptHeaderRequestCallback(responseType);
    		HttpMessageConverterExtractor<T> responseExtractor =
    				new HttpMessageConverterExtractor<T>(responseType, getMessageConverters(), logger);
    		return execute(url, HttpMethod.GET, requestCallback, responseExtractor);
    	}
    

    getForObject都调用了execute方法(其实RestTemplate的其它http请求方法调用的也是execute方法)

    	@Override
    	public <T> T execute(String url, HttpMethod method, RequestCallback requestCallback,
    			ResponseExtractor<T> responseExtractor, Object... uriVariables) throws RestClientException {
    
    		URI expanded = getUriTemplateHandler().expand(url, uriVariables);
    		return doExecute(expanded, method, requestCallback, responseExtractor);
    	}
    
    	@Override
    	public <T> T execute(String url, HttpMethod method, RequestCallback requestCallback,
    			ResponseExtractor<T> responseExtractor, Map<String, ?> uriVariables) throws RestClientException {
    
    		URI expanded = getUriTemplateHandler().expand(url, uriVariables);
    		return doExecute(expanded, method, requestCallback, responseExtractor);
    	}
    
    	@Override
    	public <T> T execute(URI url, HttpMethod method, RequestCallback requestCallback,
    			ResponseExtractor<T> responseExtractor) throws RestClientException {
    
    		return doExecute(url, method, requestCallback, responseExtractor);
    	}
    

    所有execute方法都调用了同一个doExecute方法

    	protected <T> T doExecute(URI url, HttpMethod method, RequestCallback requestCallback,
    			ResponseExtractor<T> responseExtractor) throws RestClientException {
    
    		Assert.notNull(url, "'url' must not be null");
    		Assert.notNull(method, "'method' must not be null");
    		ClientHttpResponse response = null;
    		try {
    			ClientHttpRequest request = createRequest(url, method);
    			if (requestCallback != null) {
    				requestCallback.doWithRequest(request);
    			}
    			response = request.execute();
    			handleResponse(url, method, response);
    			if (responseExtractor != null) {
    				return responseExtractor.extractData(response);
    			}
    			else {
    				return null;
    			}
    		}
    		catch (IOException ex) {
    			String resource = url.toString();
    			String query = url.getRawQuery();
    			resource = (query != null ? resource.substring(0, resource.indexOf('?')) : resource);
    			throw new ResourceAccessException("I/O error on " + method.name() +
    					" request for "" + resource + "": " + ex.getMessage(), ex);
    		}
    		finally {
    			if (response != null) {
    				response.close();
    			}
    		}
    	}
    

    doExecute方法创建了请求,然后执行,处理异常,最后关闭。可以看到关闭操作放在finally中,任何情况都会执行到,除非返回的response为null。

    InterceptingClientHttpRequest

    进入到request.execute()方法中,对应抽象类org.springframework.http.client.AbstractClientHttpRequest的execute方法

    	@Override
    	public final ClientHttpResponse execute() throws IOException {
    		assertNotExecuted();
    		ClientHttpResponse result = executeInternal(this.headers);
    		this.executed = true;
    		return result;
    	}
    

    调用内部方法executeInternal,executeInternal方法是一个抽象方法,由子类实现(restTemplate内部的http调用实现方式有多种)。进入executeInternal方法,到达抽象类
    org.springframework.http.client.AbstractBufferingClientHttpRequest

    	protected ClientHttpResponse executeInternal(HttpHeaders headers) throws IOException {
    		byte[] bytes = this.bufferedOutput.toByteArray();
    		if (headers.getContentLength() < 0) {
    			headers.setContentLength(bytes.length);
    		}
    		ClientHttpResponse result = executeInternal(headers, bytes);
    		this.bufferedOutput = null;
    		return result;
    	}
    

    缓充请求body数据,调用内部方法executeInternal

    ClientHttpResponse result = executeInternal(headers, bytes);
    

    executeInternal方法中调用另一个executeInternal方法,它也是一个抽象方法

    进入executeInternal方法,此方法由org.springframework.http.client.AbstractBufferingClientHttpRequest的子类org.springframework.http.client.InterceptingClientHttpRequest实现

    	protected final ClientHttpResponse executeInternal(HttpHeaders headers, byte[] bufferedOutput) throws IOException {
    		InterceptingRequestExecution requestExecution = new InterceptingRequestExecution();
    		return requestExecution.execute(this, bufferedOutput);
    	}
    

    实例化了一个带拦截器的请求执行对象InterceptingRequestExecution

    		public ClientHttpResponse execute(HttpRequest request, final byte[] body) throws IOException {
                  // 如果有拦截器,则执行拦截器并返回结果
    			if (this.iterator.hasNext()) {
    				ClientHttpRequestInterceptor nextInterceptor = this.iterator.next();
    				return nextInterceptor.intercept(request, body, this);
    			}
    			else {
                   // 如果没有拦截器,则通过requestFactory创建request对象并执行
    				ClientHttpRequest delegate = requestFactory.createRequest(request.getURI(), request.getMethod());
    				for (Map.Entry<String, List<String>> entry : request.getHeaders().entrySet()) {
    					List<String> values = entry.getValue();
    					for (String value : values) {
    						delegate.getHeaders().add(entry.getKey(), value);
    					}
    				}
    				if (body.length > 0) {
    					if (delegate instanceof StreamingHttpOutputMessage) {
    						StreamingHttpOutputMessage streamingOutputMessage = (StreamingHttpOutputMessage) delegate;
    						streamingOutputMessage.setBody(new StreamingHttpOutputMessage.Body() {
    							@Override
    							public void writeTo(final OutputStream outputStream) throws IOException {
    								StreamUtils.copy(body, outputStream);
    							}
    						});
    					 }
    					else {
    						StreamUtils.copy(body, delegate.getBody());
    					}
    				}
    				return delegate.execute();
    			}
    		}
    

    InterceptingClientHttpRequest的execute方法,先执行拦截器,最后执行真正的请求对象(什么是真正的请求对象?见后面拦截器的设计部分)。

    看一下RestTemplate的配置:

            RestTemplateBuilder builder = new RestTemplateBuilder();
            return builder
                    .setConnectTimeout(customConfig.getRest().getConnectTimeOut())
                    .setReadTimeout(customConfig.getRest().getReadTimeout())
                    .interceptors(restTemplateLogInterceptor)
                    .errorHandler(new ThrowErrorHandler())
                    .build();
        }
    

    可以看到配置了连接超时,读超时,拦截器,和错误处理器。
    看一下拦截器的实现:

        public ClientHttpResponse intercept(HttpRequest httpRequest, byte[] bytes, ClientHttpRequestExecution clientHttpRequestExecution) throws IOException {
            // 打印访问前日志
            ClientHttpResponse execute = clientHttpRequestExecution.execute(httpRequest, bytes);
            if (如果返回码不是200) {
                // 抛出自定义运行时异常
            }
            // 打印访问后日志
            return execute;
        }
    

    可以看到当返回码不是200时,抛出异常。还记得RestTemplate中的doExecute方法吧,此处如果抛出异常,虽然会执行doExecute方法中的finally代码,但由于返回的response为null(其实是有response的),没有关闭response,所以这里不能抛出异常,如果确实想抛出异常,可以在错误处理器errorHandler中抛出,这样确保response能正常返回和关闭。

    RestTemplate源码部分解析

    如何决定使用哪一个底层http框架

    知道了原因,我们再来看一下RestTemplate在什么时候决定使用什么http框架。其实在通过RestTemplateBuilder实例化RestTemplate对象时就决定了。
    看一下RestTemplateBuilder的build方法

    	public RestTemplate build() {
    		return build(RestTemplate.class);
    	}
    	public <T extends RestTemplate> T build(Class<T> restTemplateClass) {
    		return configure(BeanUtils.instantiate(restTemplateClass));
    	}
    

    可以看到在实例化RestTemplate对象之后,进行配置。可以指定requestFactory,也可以自动探测

    	public <T extends RestTemplate> T configure(T restTemplate) {
                   // 配置requestFactory
    		configureRequestFactory(restTemplate);
            .....省略其它无关代码
    	}
    
    
    	private void configureRequestFactory(RestTemplate restTemplate) {
    		ClientHttpRequestFactory requestFactory = null;
    		if (this.requestFactory != null) {
    			requestFactory = this.requestFactory;
    		}
    		else if (this.detectRequestFactory) {
    			requestFactory = detectRequestFactory();
    		}
    		if (requestFactory != null) {
    			ClientHttpRequestFactory unwrappedRequestFactory = unwrapRequestFactoryIfNecessary(
    					requestFactory);
    			for (RequestFactoryCustomizer customizer : this.requestFactoryCustomizers) {
    				customizer.customize(unwrappedRequestFactory);
    			}
    			restTemplate.setRequestFactory(requestFactory);
    		}
    	}
    

    看一下detectRequestFactory方法

    	private ClientHttpRequestFactory detectRequestFactory() {
    		for (Map.Entry<String, String> candidate : REQUEST_FACTORY_CANDIDATES
    				.entrySet()) {
    			ClassLoader classLoader = getClass().getClassLoader();
    			if (ClassUtils.isPresent(candidate.getKey(), classLoader)) {
    				Class<?> factoryClass = ClassUtils.resolveClassName(candidate.getValue(),
    						classLoader);
    				ClientHttpRequestFactory requestFactory = (ClientHttpRequestFactory) BeanUtils
    						.instantiate(factoryClass);
    				initializeIfNecessary(requestFactory);
    				return requestFactory;
    			}
    		}
    		return new SimpleClientHttpRequestFactory();
    	}
    

    循环REQUEST_FACTORY_CANDIDATES集合,检查classpath类路径中是否存在相应的jar包,如果存在,则创建相应框架的封装类对象。如果都不存在,则返回使用JDK方式实现的RequestFactory对象。

    看一下REQUEST_FACTORY_CANDIDATES集合

    	private static final Map<String, String> REQUEST_FACTORY_CANDIDATES;
    
    	static {
    		Map<String, String> candidates = new LinkedHashMap<String, String>();
    		candidates.put("org.apache.http.client.HttpClient",
    				"org.springframework.http.client.HttpComponentsClientHttpRequestFactory");
    		candidates.put("okhttp3.OkHttpClient",
    				"org.springframework.http.client.OkHttp3ClientHttpRequestFactory");
    		candidates.put("com.squareup.okhttp.OkHttpClient",
    				"org.springframework.http.client.OkHttpClientHttpRequestFactory");
    		candidates.put("io.netty.channel.EventLoopGroup",
    				"org.springframework.http.client.Netty4ClientHttpRequestFactory");
    		REQUEST_FACTORY_CANDIDATES = Collections.unmodifiableMap(candidates);
    	}
    

    可以看到共有四种Http调用实现方式,在配置RestTemplate时可指定,并在类路径中提供相应的实现jar包。

    Request拦截器的设计

    再看一下InterceptingRequestExecution类的execute方法。

      public ClientHttpResponse execute(HttpRequest request, final byte[] body) throws IOException {
            // 如果有拦截器,则执行拦截器并返回结果
    	  if (this.iterator.hasNext()) {
    			ClientHttpRequestInterceptor nextInterceptor = this.iterator.next();
    			return nextInterceptor.intercept(request, body, this);
    		}
    		else {
             // 如果没有拦截器,则通过requestFactory创建request对象并执行
    		    ClientHttpRequest delegate = requestFactory.createRequest(request.getURI(), request.getMethod());
    			for (Map.Entry<String, List<String>> entry : request.getHeaders().entrySet()) {
    			    List<String> values = entry.getValue();
    				for (String value : values) {
    					delegate.getHeaders().add(entry.getKey(), value);
    				}
    			}
    			if (body.length > 0) {
    				if (delegate instanceof StreamingHttpOutputMessage) {
    					StreamingHttpOutputMessage streamingOutputMessage = (StreamingHttpOutputMessage) delegate;
    					streamingOutputMessage.setBody(new StreamingHttpOutputMessage.Body() {
    						@Override
    						public void writeTo(final OutputStream outputStream) throws IOException {
    							StreamUtils.copy(body, outputStream);
    						}
    					});
    			   }
    			   else {
    				StreamUtils.copy(body, delegate.getBody());
    			   }
    			}
    			return delegate.execute();
    		}
       }
    

    大家可能会有疑问,传入的对象已经是request对象了,为什么在没有拦截器时还要再创建一遍request对象呢?
    其实传入的request对象在有拦截器的时候是InterceptingClientHttpRequest对象,没有拦截器时,则直接是包装了各个http调用实现框的Request。如HttpComponentsClientHttpRequestOkHttp3ClientHttpRequest等。当有拦截器时,会执行拦截器,拦截器可以有多个,而这里 this.iterator.hasNext() 不是一个循环,为什么呢?秘密在于拦截器的intercept方法。

    ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution)
          throws IOException;
    
    

    此方法包含request,body,execution。exection类型为ClientHttpRequestExecution接口,上面的InterceptingRequestExecution便实现了此接口,这样在调用拦截器时,传入exection对象本身,然后再调一次execute方法,再判断是否仍有拦截器,如果有,再执行下一个拦截器,将所有拦截器执行完后,再生成真正的request对象,执行http调用。

    那如果没有拦截器呢?
    上面已经知道RestTemplate在实例化时会实例化RequestFactory,当发起http请求时,会执行restTemplate的doExecute方法,此方法中会创建Request,而createRequest方法中,首先会获取RequestFactory

    // org.springframework.http.client.support.HttpAccessor
    protected ClientHttpRequest createRequest(URI url, HttpMethod method) throws IOException {
       ClientHttpRequest request = getRequestFactory().createRequest(url, method);
       if (logger.isDebugEnabled()) {
          logger.debug("Created " + method.name() + " request for "" + url + """);
       }
       return request;
    }
    
    
    // org.springframework.http.client.support.InterceptingHttpAccessor
    public ClientHttpRequestFactory getRequestFactory() {
       ClientHttpRequestFactory delegate = super.getRequestFactory();
       if (!CollectionUtils.isEmpty(getInterceptors())) {
          return new InterceptingClientHttpRequestFactory(delegate, getInterceptors());
       }
       else {
          return delegate;
       }
    }
    

    看一下RestTemplate与这两个类的关系就知道调用关系了。

    而在获取到RequestFactory之后,判断有没有拦截器,如果有,则创建InterceptingClientHttpRequestFactory对象,而此RequestFactory在createRequest时,会创建InterceptingClientHttpRequest对象,这样就可以先执行拦截器,最后执行创建真正的Request对象执行http调用。

    连接获取逻辑流程图

    以HttpComponents为底层Http调用实现的逻辑流程图。

    流程图说明:

    1. RestTemplate可以根据配置来实例化对应的RequestFactory,包括apache httpComponents、OkHttp3、Netty等实现。
    2. RestTemplate与HttpComponents衔接的类是HttpClient,此类是apache httpComponents提供给用户使用,执行http调用。HttpClient是创建RequestFactory对象时通过HttpClientBuilder实例化的,在实例化HttpClient对象时,实例化了HttpClientConnectionManager和多个ClientExecChainHttpRequestExecutorHttpProcessor以及一些策略。
    3. 当发起请求时,由requestFactory实例化httpRequest,然后依次执行ClientexecChain,常用的有四种:
      • RedirectExec: 请求跳转;根据上次响应结果和跳转策略决定下次跳转的地址,默认最大执行50次跳转;
      • RetryExec:决定出现I/O错误的请求是否再次执行
      • ProtocolExec: 填充必要的http请求header,处理http响应header,更新会话状态
      • MainClientExec:请求执行链中最后一个节点;从连接池CPool中获取连接,执行请求调用,并返回请求结果;
    4. PoolingHttpClientConnectionManager用于管理连接池,包括连接池初始化,获取连接,获取连接,打开连接,释放连接,关闭连接池等操作。
    5. CPool代表连接池,但连接并不保存在CPool中;CPool中维护着三个连接状态集合:leased(租用的,即待释放的)/available(可用的)/pending(等待的),用于记录所有连接的状态;并且维护着每个Route对应的连接池RouteSpecificPool;
    6. RouteSpecificPool是连接真正存放的地方,内部同样也维护着三个连接状态集合,但只记录属于本route的连接。
      HttpComponents将连接按照route划分连接池,有利于资源隔离,使每个route请求相互不影响;

    结束语

    • 在使用框架时,特别是在增强其功能,自定义行为时,要考虑到自定义行为对框架原有流程逻辑的影响,并且最好要熟悉框架相应功能的设计意图。
    • 在与外部事物交互,包括网络,磁盘,数据库等,做到异常情况的处理,保证程序健壮性。
  • 相关阅读:
    Mysql常用sql语句(6)- limit 限制查询结果的条数
    Mysql常用sql语句(5)- as 设置别名
    Mysql常用sql语句(4)- distinct 去重数据
    Mysql常用sql语句(3)- select 查询语句基础使用
    Jenkins(8)- CentOS 7.x 通过yum安装jenkins
    Jmeter系列(11)- 并发线程组Concurrency Thread Group详解
    Jmeter系列(10)- 阶梯加压线程组Stepping Thread Group详解
    Jmeter系列(9)- jmeter插件入门篇
    Jmeter系列(8)- test plam测试计划参数详解
    Jmeter系列(6)- test plan测试计划详细讲解
  • 原文地址:https://www.cnblogs.com/mycodingworld/p/restTemplate_intercepter.html
Copyright © 2011-2022 走看看