Merge pull request #178 from longbai/enlarge_pool_size

longbai · longbai · commit 43713ccf93cf · 2015-10-09T12:27:39.000+08:00
enlarge_pool_size
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,13 @@
 #Changelog
 
+## 7.0.4.3 (2015-10-08)
+
+### 修正
+* 调大连接池，提高并发性能
+
+### 增加
+* 增加data的异步上传
+
 ## 7.0.4.2 (2015-09-22)
 
 ### 修正
diff --git a/build.gradle b/build.gradle
@@ -16,6 +16,7 @@ dependencies {
     compile group:'com.google.code.gson', name:'gson', version:'2.3.1'
     testCompile group: 'junit', name: 'junit', version: '4.11'
     //okhttp jdk 6 version, only for jdk6 test
+    testCompile fileTree(dir: 'libs', include: '*.jar')
     testRuntime fileTree(dir: 'libs', include: '*.jar')
 }
 
diff --git a/src/main/java/com/qiniu/common/Config.java b/src/main/java/com/qiniu/common/Config.java
@@ -6,7 +6,7 @@
 
 public final class Config {
 
-    public static final String VERSION = "7.0.4.1";
+    public static final String VERSION = "7.0.4.3";
     /**
      * 断点上传时的分块大小(默认的分块大小, 不允许改变)
      */
@@ -38,13 +38,13 @@ public final class Config {
      */
     public static int PUT_THRESHOLD = BLOCK_SIZE;
     /**
-     * 连接超时时间(默认10s)
+     * 连接超时时间 单位秒(默认10s)
      */
-    public static int CONNECT_TIMEOUT = 10 * 1000;
+    public static int CONNECT_TIMEOUT = 10;
     /**
-     * 回复超时时间(默认30s)
+     * 回复超时时间 单位秒(默认30s)
      */
-    public static int RESPONSE_TIMEOUT = 30 * 1000;
+    public static int RESPONSE_TIMEOUT = 30;
     /**
      * 上传失败重试次数
      */
diff --git a/src/main/java/com/qiniu/http/AsyncCallback.java b/src/main/java/com/qiniu/http/AsyncCallback.java
@@ -0,0 +1,8 @@
+package com.qiniu.http;
+
+/**
+ * Created by bailong on 15/10/8.
+ */
+public interface AsyncCallback {
+    void complete(Response r);
+}
diff --git a/src/main/java/com/qiniu/http/Client.java b/src/main/java/com/qiniu/http/Client.java
@@ -22,7 +22,13 @@ public final class Client {
     private final OkHttpClient httpClient;
 
     public Client() {
+        Dispatcher dispatcher = new Dispatcher();
+        dispatcher.setMaxRequests(64);
+        dispatcher.setMaxRequestsPerHost(16);
+        ConnectionPool connectionPool = new ConnectionPool(32, 5 * 60 * 1000);
         httpClient = new OkHttpClient();
+        httpClient.setDispatcher(dispatcher);
+        httpClient.setConnectionPool(connectionPool);
         httpClient.networkInterceptors().add(new Interceptor() {
             @Override
             public com.squareup.okhttp.Response intercept(Chain chain) throws IOException {
@@ -198,6 +204,95 @@ public void accept(String key, Object value) {
         return r;
     }
 
+    public void asyncSend(final Request.Builder requestBuilder, StringMap headers, final AsyncCallback cb) {
+        if (headers != null) {
+            headers.forEach(new StringMap.Consumer() {
+                @Override
+                public void accept(String key, Object value) {
+                    requestBuilder.header(key, value.toString());
+                }
+            });
+        }
+
+        requestBuilder.header("User-Agent", userAgent());
+        final long start = System.currentTimeMillis();
+        IpTag tag = new IpTag();
+        httpClient.newCall(requestBuilder.tag(tag).build()).enqueue(new Callback() {
+            @Override
+            public void onFailure(Request request, IOException e) {
+                e.printStackTrace();
+                long duration = (System.currentTimeMillis() - start) / 1000;
+                cb.complete(Response.createError(null, "", duration, e.getMessage()));
+            }
+
+            @Override
+            public void onResponse(com.squareup.okhttp.Response response) throws IOException {
+                long duration = (System.currentTimeMillis() - start) / 1000;
+                cb.complete(Response.create(response, "", duration));
+            }
+        });
+    }
+
+    public void asyncPost(String url, byte[] body, int offset, int size,
+                          StringMap headers, String contentType, AsyncCallback cb) {
+        RequestBody rbody;
+        if (body != null && body.length > 0) {
+            MediaType t = MediaType.parse(contentType);
+            rbody = create(t, body, offset, size);
+        } else {
+            rbody = RequestBody.create(null, new byte[0]);
+        }
+
+        Request.Builder requestBuilder = new Request.Builder().url(url).post(rbody);
+        asyncSend(requestBuilder, headers, cb);
+    }
+
+    public void asyncMultipartPost(String url,
+                                   StringMap fields,
+                                   String name,
+                                   String fileName,
+                                   byte[] fileBody,
+                                   String mimeType,
+                                   StringMap headers,
+                                   AsyncCallback cb) {
+        RequestBody file = RequestBody.create(MediaType.parse(mimeType), fileBody);
+        asyncMultipartPost(url, fields, name, fileName, file, headers, cb);
+    }
+
+    public void asyncMultipartPost(String url,
+                                   StringMap fields,
+                                   String name,
+                                   String fileName,
+                                   File fileBody,
+                                   String mimeType,
+                                   StringMap headers,
+                                   AsyncCallback cb) throws QiniuException {
+        RequestBody file = RequestBody.create(MediaType.parse(mimeType), fileBody);
+        asyncMultipartPost(url, fields, name, fileName, file, headers, cb);
+    }
+
+    private void asyncMultipartPost(String url,
+                                    StringMap fields,
+                                    String name,
+                                    String fileName,
+                                    RequestBody file,
+                                    StringMap headers,
+                                    AsyncCallback cb) {
+        final MultipartBuilder mb = new MultipartBuilder();
+        mb.addFormDataPart(name, fileName, file);
+
+        fields.forEach(new StringMap.Consumer() {
+            @Override
+            public void accept(String key, Object value) {
+                mb.addFormDataPart(key, value.toString());
+            }
+        });
+        mb.type(MediaType.parse("multipart/form-data"));
+        RequestBody body = mb.build();
+        Request.Builder requestBuilder = new Request.Builder().url(url).post(body);
+        asyncSend(requestBuilder, headers, cb);
+    }
+
     private static class IpTag {
         public String ip = null;
     }
diff --git a/src/main/java/com/qiniu/http/Response.java b/src/main/java/com/qiniu/http/Response.java
@@ -88,6 +88,33 @@ static Response create(com.squareup.okhttp.Response response, String address, do
                 address, duration, error, body);
     }
 
+    static Response createError(com.squareup.okhttp.Response response, String address, double duration, String error) {
+        if (response == null) {
+            return new Response(null, -1, "", "", "", "", duration, error, null);
+        }
+        int code = response.code();
+        String reqId = null;
+
+        byte[] body = null;
+        if (ctype(response).equals(Client.JsonMime)) {
+            reqId = response.header("X-Reqid");
+            reqId = (reqId == null) ? null : reqId.trim();
+            try {
+                body = response.body().bytes();
+                if (response.code() >= 400 && !StringUtils.isNullOrEmpty(reqId) && body != null) {
+                    ErrorBody errorBody = Json.decode(new String(body), ErrorBody.class);
+                    error = errorBody.error;
+                }
+            } catch (Exception e) {
+                if (response.code() < 300) {
+                    error = e.getMessage();
+                }
+            }
+        }
+        return new Response(response, code, reqId, response.header("X-Log"), via(response),
+                address, duration, error, body);
+    }
+
 
     private static String via(com.squareup.okhttp.Response response) {
         String via;
diff --git a/src/main/java/com/qiniu/storage/AsyncResumeUploader.java b/src/main/java/com/qiniu/storage/AsyncResumeUploader.java
@@ -0,0 +1,7 @@
+package com.qiniu.storage;
+
+/**
+ * Created by bailong on 15/10/8.
+ */
+public class AsyncResumeUploader {
+}
diff --git a/src/main/java/com/qiniu/storage/FormUploader.java b/src/main/java/com/qiniu/storage/FormUploader.java
@@ -2,6 +2,7 @@
 
 import com.qiniu.common.Config;
 import com.qiniu.common.QiniuException;
+import com.qiniu.http.AsyncCallback;
 import com.qiniu.http.Client;
 import com.qiniu.http.Response;
 import com.qiniu.util.Crc32;
@@ -53,6 +54,27 @@ Response upload() throws QiniuException {
         return client.multipartPost(Config.zone.upHost, params, "file", fileName, file, mime, new StringMap());
     }
 
+    void asyncUpload(final UpCompletionHandler handler) throws IOException {
+        buildParams();
+        if (data != null) {
+            client.asyncMultipartPost(Config.zone.upHost, params, "file", fileName,
+                    data, mime, new StringMap(), new AsyncCallback() {
+                        @Override
+                        public void complete(Response r) {
+                            handler.complete(key, r);
+                        }
+                    });
+            return;
+        }
+        client.asyncMultipartPost(Config.zone.upHost, params, "file", fileName,
+                file, mime, new StringMap(), new AsyncCallback() {
+                    @Override
+                    public void complete(Response r) {
+                        handler.complete(key, r);
+                    }
+                });
+    }
+
     private void buildParams() throws QiniuException {
         params.put("token", token);
         if (key == null) {
diff --git a/src/main/java/com/qiniu/storage/UpCompletionHandler.java b/src/main/java/com/qiniu/storage/UpCompletionHandler.java
@@ -0,0 +1,10 @@
+package com.qiniu.storage;
+
+import com.qiniu.http.Response;
+
+/**
+ * Created by bailong on 15/10/8.
+ */
+public interface UpCompletionHandler {
+    void complete(String key, Response r);
+}
diff --git a/src/main/java/com/qiniu/storage/UploadManager.java b/src/main/java/com/qiniu/storage/UploadManager.java
@@ -7,6 +7,7 @@
 import com.qiniu.util.StringMap;
 
 import java.io.File;
+import java.io.IOException;
 
 /**
  * 七牛文件上传管理器
@@ -191,4 +192,15 @@ public Response put(File file, String key, String token, StringMap params,
                 params, mime, recorder, recorderKey);
         return uploader.upload();
     }
+
+
+    public void asyncPut(final byte[] data, final String key, final String token, StringMap params,
+                         String mime, boolean checkCrc, UpCompletionHandler handler) throws IOException {
+        checkArgs(key, data, null, token);
+        if (mime == null) {
+            mime = Client.DefaultMime;
+        }
+        params = filterParam(params);
+        new FormUploader(client, token, key, data, params, mime, checkCrc).asyncUpload(handler);
+    }
 }
diff --git a/src/test/java/com/qiniu/processing/PfopTest.java b/src/test/java/com/qiniu/processing/PfopTest.java
@@ -5,7 +5,6 @@
 import com.qiniu.http.Response;
 import com.qiniu.util.Auth;
 import com.qiniu.util.StringMap;
-
 import com.qiniu.util.UrlSafeBase64;
 import org.junit.Test;
 
diff --git a/src/test/java/com/qiniu/storage/FormUploadTest.java b/src/test/java/com/qiniu/storage/FormUploadTest.java
@@ -9,6 +9,8 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.*;
 
@@ -128,4 +130,41 @@ public void testFile() {
         }
         TempFile.remove(f);
     }
+
+    @Test
+    public void testAsync() {
+        final String expectKey = "你好?&=\r\n";
+        StringMap params = new StringMap().put("x:foo", "foo_val");
+
+        String token = TestConfig.testAuth.uploadToken(TestConfig.bucket, expectKey);
+        final CountDownLatch signal = new CountDownLatch(1);
+        Response r = null;
+        try {
+            uploadManager.asyncPut("hello".getBytes(), expectKey, token, params,
+                    null, false, new UpCompletionHandler() {
+                @Override
+                public void complete(String key, Response r) {
+                    signal.countDown();
+                    StringMap map = null;
+                    try {
+                        map = r.jsonToMap();
+                    } catch (QiniuException e) {
+                        e.printStackTrace();
+                        fail();
+                    }
+                    assertEquals(200, r.statusCode);
+                    assert map != null;
+                    assertEquals("Fqr0xh3cxeii2r7eDztILNmuqUNN", map.get("hash"));
+                    assertEquals(expectKey, map.get("key"));
+                }
+            });
+        } catch (IOException e) {
+            fail();
+        }
+        try {
+            signal.await(120, TimeUnit.SECONDS); // wait for callback
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
 }
diff --git a/src/test/java/com/qiniu/storage/RecordUploadTest.java b/src/test/java/com/qiniu/storage/RecordUploadTest.java

Original file line number	Diff line number	Diff line change
`@@ -16,6 +16,7 @@ dependencies {`
`16`	`16`	`compile group:'com.google.code.gson', name:'gson', version:'2.3.1'`
`17`	`17`	`testCompile group: 'junit', name: 'junit', version: '4.11'`
`18`	`18`	`//okhttp jdk 6 version, only for jdk6 test`
	`19`	`+ testCompile fileTree(dir: 'libs', include: '*.jar')`
`19`	`20`	`testRuntime fileTree(dir: 'libs', include: '*.jar')`
`20`	`21`	`}`
`21`	`22`