Using Java Apache PoolingClientConnectionManager l

Question

My web application run jobs in night!And meet the problem!It used a lot of memory!
I use the command to find which function occupy the java resource!
It is result:

        [tomcat@uhzd006525 ~]$ jstack 2365 |grep 93f -A 30
            - parking to wait for  <0x00000007eac93f68> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
            at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
            at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
            at org.apache.http.pool.PoolEntryFuture.await(PoolEntryFuture.java:131)
            at org.apache.http.pool.AbstractConnPool.getPoolEntryBlocking(AbstractConnPool.java:281)
            at org.apache.http.pool.AbstractConnPool.access$000(AbstractConnPool.java:62)
            at org.apache.http.pool.AbstractConnPool$2.getPoolEntry(AbstractConnPool.java:176)
            at org.apache.http.pool.AbstractConnPool$2.getPoolEntry(AbstractConnPool.java:172)
            at org.apache.http.pool.PoolEntryFuture.get(PoolEntryFuture.java:100)
            at org.apache.http.impl.conn.PoolingClientConnectionManager.leaseConnection(PoolingClientConnectionManager.java:212)
            at org.apache.http.impl.conn.PoolingClientConnectionManager$1.getConnection(PoolingClientConnectionManager.java:199)
            at org.apache.http.impl.client.DefaultRequestDirector.execute(DefaultRequestDirector.java:456)
            at org.apache.http.impl.client.AbstractHttpClient.execute(AbstractHttpClient.java:906)
            at org.apache.http.impl.client.AbstractHttpClient.execute(AbstractHttpClient.java:805)
            at com.trendata.spider.PageGetter.getPageContent(PageGetter.java:262)
            at com.trendata.spider.PageGetter.getTaobaoContent(PageGetter.java:376)
            at com.trendata.taobao.MbpBkDataCreator.getBkMbpValue(MbpBkDataCreator.java:48)
            at com.trendata.taobao.MbpBkDataCreator.getIntoStores(MbpBkDataCreator.java:106)
            at com.trendata.service.impl.OddJobsServiceImpl.getLast7DaysIntoStore(OddJobsServiceImpl.java:448)
            at sun.reflect.GeneratedMethodAccessor205.invoke(Unknown Source)
            at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
            at java.lang.reflect.Method.invoke(Method.java:601)
            at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:319)
            at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:183)
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:150)
            at com.trendata.service.interceptor.MethodCacheInterceptor.invoke(MethodCacheInterceptor.java:32)
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)
            at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:202)
            at $Proxy144.getLast7DaysIntoStore(Unknown Source)
            at sun.reflect.GeneratedMethodAccessor205.invoke(Unknown Source)
            at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    --
    "GC task thread#0 (ParallelGC)" prio=10 tid=0x0000000000d46000 nid=0x93f runnable

    "GC task thread#1 (ParallelGC)" prio=10 tid=0x0000000000d48000 nid=0x940 runnable

    "VM Periodic Task Thread" prio=10 tid=0x00002aaabc059800 nid=0x94b waiting on condition

    JNI global references: 342






    "Thread-114" prio=10 tid=0x000000002281e800 nid=0xf4a waiting on condition [0x000000004c7d3000]
       java.lang.Thread.State: WAITING (parking)
            at sun.misc.Unsafe.park(Native Method)
            - parking to wait for  <0x0000000790f5dc40> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
            at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
            at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
            at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
            at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1043)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1103)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
            at java.lang.Thread.run(Thread.java:722)

    "Thread-113" prio=10 tid=0x0000000022624800 nid=0xf48 waiting on condition [0x000000004c6d2000]
       java.lang.Thread.State: WAITING (parking)
            at sun.misc.Unsafe.park(Native Method)
            - parking to wait for  <0x0000000790f5dc40> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
            at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
            at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
            at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
            at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1043)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1103)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
            at java.lang.Thread.run(Thread.java:722)

    "Thread-112" prio=10 tid=0x00000000225d9800 nid=0xf37 waiting on condition [0x000000004c5d1000]
       java.lang.Thread.State: WAITING (parking)
            at sun.misc.Unsafe.park(Native Method)
            - parking to wait for  <0x0000000790f5dc40> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
            at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
            at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
            at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
    --
    "GC task thread#0 (ParallelGC)" prio=10 tid=0x000000001f287000 nid=0x7d30 runnable

    "GC task thread#1 (ParallelGC)" prio=10 tid=0x000000001f288800 nid=0x7d31 runnable

    "VM Periodic Task Thread" prio=10 tid=0x000000001f3d1000 nid=0x7d39 waiting on condition

    JNI global references: 579

This is the Code that happened problem:

    public class PageGetter {
    private static final Log log = LogFactory.getLog("serviceLogger");
    private static SyncBasicHttpParams httpParams = null;
    private HttpClient httpClient = null;
    private static PoolingClientConnectionManager connectionManager = null;
    private static Integer errorSleepTime = new Integer(PropertyGetter.getInstance().getProperty("errorpagesleeptime"));

    public String getFinalRedirectURL(String url) throws ClientProtocolException, IOException {
        if (url == null) {
            return null;
        }
        if (!url.trim().startsWith("http://")) {
            url = "http://" + url;
        }
        HttpContext localContext = new BasicHttpContext();
        HttpGet httpget = new HttpGet(url);
        if (connectionManager == null || httpParams == null || this.httpClient == null) {
            this.initHttpManager();
        }
        this.httpClient.execute(httpget, localContext);
        HttpHost target = (HttpHost) localContext.getAttribute(ExecutionContext.HTTP_TARGET_HOST);
        String finalURL = target.toString();
        httpget.abort();
        return finalURL;
    }

    private void initHttpManager() {
        // Create and initialize HTTP parameters
        if (httpParams == null) {
            httpParams = new SyncBasicHttpParams();
            httpParams.setIntParameter(CoreConnectionPNames.SO_TIMEOUT, 60000);
            httpParams.setParameter(HTTP.CONTENT_ENCODING, "GBK");
            HttpProtocolParams.setVersion(httpParams, HttpVersion.HTTP_1_1);
        }
        if (connectionManager == null) {
            SchemeRegistry schemeRegistry = new SchemeRegistry();
            schemeRegistry.register(new Scheme("http", 80, PlainSocketFactory.getSocketFactory()));
            try {
                //Security Socket
                SSLContext ctx = SSLContext.getInstance("TLS");
                X509TrustManager tm = new X509TrustManager() {
                    @Override
                    public X509Certificate[] getAcceptedIssuers() {
                        return null;
                    }

                    @Override
                    public void checkClientTrusted(X509Certificate[] arg0, String arg1) throws CertificateException {
                    }

                    @Override
                    public void checkServerTrusted(X509Certificate[] arg0, String arg1) throws CertificateException {
                    }
                };
                ctx.init(null, new TrustManager[] { tm }, null);
                SSLSocketFactory ssf = new SSLSocketFactory(ctx, SSLSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
                schemeRegistry.register(new Scheme("https", 443, ssf));
            } catch (Exception e) {
                log.error("Fail to create connection manager " + e);
            }
            connectionManager = new PoolingClientConnectionManager(schemeRegistry);
        }
        this.httpClient = new DefaultHttpClient(connectionManager, httpParams);
        this.httpClient.getParams().setParameter(ClientPNames.ALLOW_CIRCULAR_REDIRECTS, true);
        HttpClientParams.setCookiePolicy(this.httpClient.getParams(), CookiePolicy.BROWSER_COMPATIBILITY);
    }

    public String getPostPageContent(String url, List<NameValuePair> headers, List<NameValuePair> params) {
        String strPage = null;
        if (url == null) {
            return null;
        }
        if (!url.trim().startsWith("http")) {
            url = "http://" + url;
        }
        HttpPost httpPost = new HttpPost(url);
        HttpEntity entity = null;
        try {
            if (params != null) {
                httpPost.setEntity(new UrlEncodedFormEntity(params, "UTF-8"));
            }
            if (headers != null) {
                Iterator<NameValuePair> itrHeaders = headers.iterator();
                while (itrHeaders.hasNext()) {
                    NameValuePair header = itrHeaders.next();
                    httpPost.setHeader(header.getName(), header.getValue());
                }
            }
            HttpContext localContext = new BasicHttpContext();
            if (connectionManager == null || httpParams == null || this.httpClient == null) {
                this.initHttpManager();
            }
            HttpResponse response = this.httpClient.execute(httpPost, localContext);
            // get the response body as an array of bytes
            entity = response.getEntity();
            if (entity != null) {
                strPage = EntityUtils.toString(entity);
            }
        } catch (Exception e) {
            connectionManager = null;
            httpParams = null;
            this.httpClient = null;
            this.errorHandler(url, e);
        } finally {
            httpPost.abort();

            if (entity != null) {
                try {
                    EntityUtils.consume(entity);
                } catch (Exception e2) {
                    e2.printStackTrace();
                    log.error("Error when consume entity ", e2);
                }
            }
        }
        if (strPage != null) {
            Pattern pattern = Pattern.compile("\\s");
            Matcher matcher = pattern.matcher(strPage);
            strPage = matcher.replaceAll(" ");
            strPage = strPage.replaceAll("////", "");
        }
        return strPage;
    }

    private String getURL(String url, List<NameValuePair> headers) {
        if (headers == null) {
            return url;
        }
        Map<String, String> map = new HashMap<String, String>();
        if (url.contains("?")) {
            String str = url.substring(url.indexOf("?") + 1);
            url = url.substring(0, url.indexOf("?"));
            String[] strs = str.split("&");
            for (int i = 0; strs != null && i < strs.length; i++) {
                if (strs[i].contains("=")) {
                    String key = strs[i].substring(0, strs[i].indexOf("="));
                    String value = strs[i].substring(strs[i].indexOf("=") + 1);
                    map.put(key, value);
                }
            }
        }
        Iterator<NameValuePair> itrHeaders = headers.iterator();
        while (itrHeaders.hasNext()) {
            NameValuePair header = itrHeaders.next();
            if (header != null && header.getName() != null && header.getValue() != null) {
                try {
                    map.put(URLEncoder.encode(header.getName(), "UTF-8"), URLEncoder.encode(header.getValue(), "UTF-8"));
                } catch (Exception e) {
                    log.error("Can not parse parameter " + e);
                }
            }
        }
        if (map.keySet() == null) {
            return url;
        }
        Iterator<String> itr = map.keySet().iterator();
        String parameters = "";
        while (itr.hasNext()) {
            String key = itr.next();
            if (key != null && key.length() > 0) {
                parameters = parameters + key + "=" + map.get(key) + "&";
            }
        }
        if (parameters.length() > 0) {
            url = url + "?" + parameters;
        }
        return url;
    }

    private void errorHandler(String url, Exception e) {
        try {
            Thread.sleep(errorSleepTime);
            log.error("无法打开链接" + url + " sleep " + errorSleepTime + e);
        } catch (Exception e1) {
            log.error("Error sleeping " + errorSleepTime + e1);
        }
    }

    /**
     * Get page as string from URL
     * 
     * @param url
     * @return
     */
    public String getPageContent(String url, List<NameValuePair> headers) {
        url = StringFilter.replaceUnicode(url);
        String strPage = null;
        if (url == null) {
            return null;
        }
        if (!url.trim().startsWith("http")) {
            url = "http://" + url;
        }
        if (url.endsWith("/")) {
            url = url.substring(0, url.lastIndexOf("/"));
        }
        HttpGet httpGet = null;
        HttpEntity entity = null;
        try {
            if (connectionManager == null || httpParams == null || this.httpClient == null) {
                this.initHttpManager();
            }
            url = this.getURL(url, headers);
            httpGet = new HttpGet(url);
            HttpContext localContext = new BasicHttpContext();
            HttpResponse response = this.httpClient.execute(httpGet, localContext);
            entity = response.getEntity();
            if (entity != null) {
                String charSet = ContentType.getOrDefault(entity).getCharset().toString();
                if (charSet == null) {
                    charSet = "UTF-8";
                }
                strPage = EntityUtils.toString(entity, charSet);
                strPage = strPage.replaceAll("\r", "");
                strPage = strPage.replaceAll("\n", "");
                //空格
                strPage = strPage.replaceAll("&nbsp;", "");
            }
        } catch (Exception e) {
            connectionManager = null;
            httpParams = null;
            this.httpClient = null;
            this.errorHandler(url, e);
        } finally {
            try {
                httpGet.abort();
            } catch (Exception e) {
                log.error("Error when httpclient aborting " + e);
            }

            if (entity != null) {
                try {
                    EntityUtils.consume(entity);
                } catch (Exception e2) {
                    e2.printStackTrace();
                    log.error("Error when consume entity ", e2);
                }
            }
        }
        return strPage;
    }

    public String getPageContentSms(String url, List<NameValuePair> headers) {
        url = StringFilter.replaceUnicode(url);
        String strPage = null;
        if (url == null) {
            return null;
        }
        if (!url.trim().startsWith("http")) {
            url = "http://" + url;
        }
        if (url.endsWith("/")) {
            url = url.substring(0, url.lastIndexOf("/"));
        }
        HttpGet httpGet = null;
        HttpEntity entity = null;
        try {
            if (connectionManager == null || httpParams == null || this.httpClient == null) {
                this.initHttpManager();
            }
            httpGet = new HttpGet(url);
            HttpContext localContext = new BasicHttpContext();
            HttpResponse response = this.httpClient.execute(httpGet, localContext);
            entity = response.getEntity();
            if (entity != null) {
                String charSet = ContentType.getOrDefault(entity).getCharset().toString();
                if (charSet == null) {
                    charSet = "UTF-8";
                }
                strPage = EntityUtils.toString(entity, charSet);
                strPage = strPage.replaceAll("\r", "");
                strPage = strPage.replaceAll("\n", "");
                strPage = strPage.replaceAll("&nbsp;", "");
            }
        } catch (Exception e) {
            connectionManager = null;
            httpParams = null;
            this.httpClient = null;
            this.errorHandler(url, e);
        } finally {
            try {
                httpGet.abort();
            } catch (Exception e) {
                log.error("Error when httpclient aborting " + e);
            }

            if (entity != null) {
                try {
                    EntityUtils.consume(entity);
                } catch (Exception e2) {
                    e2.printStackTrace();
                    log.error("Error when consume entity ", e2);
                }
            }
        }
        return strPage;
    }

    public List<NameValuePair> getDefaultHeaders() {
        List<NameValuePair> headers = new ArrayList<NameValuePair>();
        BasicNameValuePair nameValuePair = new BasicNameValuePair("Accept",
                "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8");
        headers.add(nameValuePair);
        nameValuePair = new BasicNameValuePair("Accept-Charset", "GBK,utf-8;q=0.7,*;q=0.3");
        headers.add(nameValuePair);
        nameValuePair = new BasicNameValuePair("Accept-Encoding", "GBK,utf-8;q=0.7,*;q=0.3");
        headers.add(nameValuePair);
        nameValuePair = new BasicNameValuePair("Accept-Language", "zh-CN,zh;q=0.8");
        headers.add(nameValuePair);
        nameValuePair = new BasicNameValuePair("Cache-Control", "max-age=0");
        headers.add(nameValuePair);
        nameValuePair = new BasicNameValuePair("Connection", "keep-alive");
        headers.add(nameValuePair);
        nameValuePair = new BasicNameValuePair("User-Agent",
                "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/535.19 (KHTML, like Gecko) Chrome/18.0.1025.162 Safari/535.19");
        headers.add(nameValuePair);
        return headers;
    }

    public String getTaobaoContent(List<NameValuePair> params) {
        return this.getPageContent(Constant.TAOBAO_API_URL, params);
    }

Answer 1

Are you using an Oracle JRE older than 1.06_21? We found a deadlock bug that produced that issue. If you're using a 32-bit Windows, you can get a memory error when you've actually run out of kernel space for your excessive threads for such an issue.

Answer 2

We were seeing very similar issue when using HttpClient with Jersey. Threads were blocked waiting for connection.

In our case the connection pool got depleted when we encountered 40x errors in our GET requests.

The reason was that the connections were never freed back to the pool unless we actually read the response body ( which we did not in the cases when we got client error codes). In your source code it seems you just throw the ClientProtocolException without reading the response.. it might be similar..