Merge pull request #231 from tencentyun/dev/checkPointUpload

update IdleConnectionMonitor

Author: Dzkol

pom.xml

@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.qcloud</groupId>
     <artifactId>cos_api</artifactId>
-    <version>5.6.240</version>
+    <version>5.6.240.2</version>
     <packaging>jar</packaging>
     <name>cos-java-sdk</name>
     <description>java sdk for qcloud cos</description>

src/main/java/com/qcloud/cos/ClientConfig.java

@@ -133,6 +133,10 @@ public class ClientConfig {
 
     private boolean isRedirectsEnabled = false;
 
+    private boolean useConnectionMonitor = false;
+
+    private long connectionMaxIdleMillis = 60 * 1000;
+
     // 不传入region 用于后续调用List Buckets(获取所有的bucket信息)
     public ClientConfig() {
         super();
@@ -477,4 +481,20 @@ public boolean isRedirectsEnabled() {
     public void setRedirectsEnabled(boolean redirectsEnabled) {
         isRedirectsEnabled = redirectsEnabled;
     }
+
+    public boolean isUseConnectionMonitor() {
+        return useConnectionMonitor;
+    }
+
+    public void setUseConnectionMonitor(boolean isUseConnectionMonitor) {
+        useConnectionMonitor = isUseConnectionMonitor;
+    }
+
+    public long getConnectionMaxIdleMillis() {
+        return connectionMaxIdleMillis;
+    }
+
+    public void setConnectionMaxIdleMillis(long connectionMaxIdleMillis) {
+        this.connectionMaxIdleMillis = connectionMaxIdleMillis;
+    }
 }

src/main/java/com/qcloud/cos/http/DefaultCosHttpClient.java

@@ -119,7 +119,7 @@ public class DefaultCosHttpClient implements CosHttpClient {
     private RequestConfig requestConfig;
     protected HttpClient httpClient;
     private PoolingHttpClientConnectionManager connectionManager;
-    private IdleConnectionMonitorThread idleConnectionMonitor;
+    private IdleConnectionMonitorThread idleConnectionMonitorThread;
     private int maxErrorRetry;
     private RetryPolicy retryPolicy;
     private BackoffStrategy backoffStrategy;
@@ -208,10 +208,14 @@ private void initHttpClient() {
                         .setSocketTimeout(this.clientConfig.getSocketTimeout())
                         .setRedirectsEnabled(this.clientConfig.isRedirectsEnabled())
                         .build();
-        this.idleConnectionMonitor = new IdleConnectionMonitorThread(this.connectionManager);
-        this.idleConnectionMonitor.setIdleAliveMS(this.clientConfig.getIdleConnectionAlive());
-        this.idleConnectionMonitor.setDaemon(true);
-        this.idleConnectionMonitor.start();
+        if (clientConfig.isUseConnectionMonitor()) {
+            IdleConnectionMonitor.registerConnectionManager(this.connectionManager, clientConfig.getConnectionMaxIdleMillis());
+        } else {
+            this.idleConnectionMonitorThread = new IdleConnectionMonitorThread(this.connectionManager);
+            this.idleConnectionMonitorThread.setIdleAliveMS(this.clientConfig.getIdleConnectionAlive());
+            this.idleConnectionMonitorThread.setDaemon(true);
+            this.idleConnectionMonitorThread.start();
+        }
     }
 
     @Override
@@ -231,7 +235,12 @@ public void shutdown() {
             log.info(trace.toString());
         }
         cosHttpClientTimer.shutdown();
-        this.idleConnectionMonitor.shutdown();
+        if (clientConfig.isUseConnectionMonitor()) {
+            IdleConnectionMonitor.removeConnectionManager(this.connectionManager);
+            this.connectionManager.shutdown();
+        } else {
+            this.idleConnectionMonitorThread.shutdown();
+        }
     }
 
     // 因为Apache HTTP库自带的URL Encode对一些特殊字符如*等不进行转换, 和COS HTTP服务的URL Encode标准不一致

src/main/java/com/qcloud/cos/http/IdleConnectionMonitor.java

@@ -0,0 +1,89 @@
+package com.qcloud.cos.http;
+
+import org.apache.http.conn.HttpClientConnectionManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+public class IdleConnectionMonitor extends Thread {
+    private static final Logger log = LoggerFactory.getLogger(IdleConnectionMonitor.class);
+    private static final Map<HttpClientConnectionManager, Long> connectionManagers = new ConcurrentHashMap<HttpClientConnectionManager, Long>();
+    private static final int PERIOD_MILLISECONDS = 1000 * 60;
+    private static volatile IdleConnectionMonitor instance;
+
+    private volatile boolean shuttingDown;
+
+    private IdleConnectionMonitor() {
+        super("cos-java-sdk-http-connection-monitor");
+        setDaemon(true);
+    }
+
+    private void markShuttingDown() {
+        shuttingDown = true;
+    }
+
+    public static boolean registerConnectionManager(HttpClientConnectionManager connectionManager, long maxIdleInMs) {
+        if (instance == null) {
+            synchronized (IdleConnectionMonitor.class) {
+                if (instance == null) {
+                    instance = new IdleConnectionMonitor();
+                    instance.start();
+                }
+            }
+        }
+        return connectionManagers.put(connectionManager, maxIdleInMs) == null;
+    }
+
+    public static boolean removeConnectionManager(HttpClientConnectionManager connectionManager) {
+        boolean wasRemoved = connectionManagers.remove(connectionManager) != null;
+        if (connectionManagers.isEmpty()) {
+            shutdown();
+        }
+        return wasRemoved;
+    }
+
+    public static List<HttpClientConnectionManager> getRegisteredConnectionManagers() {
+        return new ArrayList<HttpClientConnectionManager>(connectionManagers.keySet());
+    }
+
+    public static synchronized boolean shutdown() {
+        if (instance != null) {
+            instance.markShuttingDown();
+            instance.interrupt();
+            connectionManagers.clear();
+            instance = null;
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public void run() {
+        while (!shuttingDown) {
+            try {
+                for (Map.Entry<HttpClientConnectionManager, Long> entry : connectionManagers.entrySet()) {
+                    // When we release connections, the connection manager leaves them
+                    // open so they can be reused.  We want to close out any idle
+                    // connections so that they don't sit around in CLOSE_WAIT.
+                    try {
+                        entry.getKey().closeExpiredConnections();
+                        entry.getKey().closeIdleConnections(entry.getValue(), TimeUnit.MILLISECONDS);
+                    } catch (Exception e) {
+                        log.warn("Unable to closeExpiredConnections and closeIdleConnections, ", e);
+                    }
+                }
+
+                Thread.sleep(PERIOD_MILLISECONDS);
+            } catch (Throwable t) {
+                log.error("error occurred when closeExpiredConnections and closeIdleConnections, err:", t);
+            }
+        }
+
+        log.debug("Shutting down reaper thread.");
+    }
+}