Java多線程下載初試

本文轉載自查看原文 2019-05-27 17:06 609

一、服務端/客戶端代碼的實現

服務端配置config

 1 @ConfigurationProperties("storage")
 2 public class StorageProperties {
 3     private String location = "D:\\idea_project\\upload\\src\\main\\resources\\upload-files";
 4 
 5     public String getLocation() {
 6         return location;
 7     }
 8 
 9     public void setLocation(String location) {
10         this.location = location;
11     }
12 }

服務端Controller

1 @GetMapping("/files/{filename:.+}")
2     @ResponseBody
3     public ResponseEntity<Resource> serveFile(@PathVariable String filename) {
4         Resource file = storageService.loadAsResource(filename);
5         return ResponseEntity.ok().header(HttpHeaders.CONTENT_DISPOSITION,
6                 "attachment; filename=\"" + file.getFilename() + "\"").body(file);
7     }

服務端Service

1 Path load(String filename);
2 
3 Resource loadAsResource(String filename);

 1 package org.wlgzs.upload.service.impl;
 2 
 3 import org.springframework.beans.factory.annotation.Autowired;
 4 import org.springframework.core.io.Resource;
 5 import org.springframework.core.io.UrlResource;
 6 import org.springframework.stereotype.Service;
 7 import org.springframework.util.FileSystemUtils;
 8 import org.springframework.util.StringUtils;
 9 import org.springframework.web.multipart.MultipartFile;
10 import org.wlgzs.upload.config.StorageProperties;
11 import org.wlgzs.upload.service.StorageService;
12 
13 import java.io.IOException;
14 import java.net.MalformedURLException;
15 import java.nio.file.Files;
16 import java.nio.file.Path;
17 import java.nio.file.Paths;
18 import java.nio.file.StandardCopyOption;
19 import java.util.stream.Stream;
20 
21 /**
22  * @author zsh
23  * @company wlgzs
24  * @create 2018-12-15 16:16
25  * @Describe
26  */
27 
28 @Service
29 public class FileSystemStorageService implements StorageService {
30 
31     private final Path rootLocation;
32 
33     @Autowired
34     public FileSystemStorageService(StorageProperties properties) {
35         this.rootLocation = Paths.get(properties.getLocation());
36     }
37 
38     @Override
39     public Path load(String filename) {
40         return rootLocation.resolve(filename);
41     }
42 
43     @Override
44     public Resource loadAsResource(String filename) {
45         try {
46             Path file = load(filename);
47             Resource resource = new UrlResource(file.toUri());
48             if (resource.exists() || resource.isReadable()) {
49                 return resource;
50             }
51             else {
52                 System.out.println("Could not read file: " + filename);
53                 //throw new StorageFileNotFoundException("Could not read file: " + filename);
54 
55             }
56         }
57         catch (MalformedURLException e) {
58             System.out.println("Could not read file: " + filename);
59             //throw new StorageFileNotFoundException("Could not read file: " + filename, e);
60         }
61         return null;
62     }
63 
64 }

服務端目錄結構

客戶端Main類

 1 import java.util.Scanner;
 2 import java.util.concurrent.TimeUnit;
 3 
 4 /**
 5  * @author zsh
 6  * @site www.qqzsh.top
 7  * @company wlgzs
 8  * @create 2019-05-27 9:03
 9  * @description 主線程啟動入口
10  */
11 public class Main {
12     public static void main(String[] args) {
13         Scanner scanner = new Scanner(System.in);
14         System.out.println("請輸入下載文件的地址，按ENTER結束");
15         String downpath = scanner.nextLine();
16         System.out.println("下載的文件名及路徑為："+ MultiPartDownLoad.downLoad(downpath));
17         try {
18             System.out.println("下載完成，本窗口5s之后自動關閉");
19             TimeUnit.SECONDS.sleep(5);
20         } catch (InterruptedException e) {
21             e.printStackTrace();
22         }
23         System.exit(0);
24     }
25 }

客戶端線程池Constans類

 1 import java.util.concurrent.*;
 2 
 3 /**
 4  * @author zsh
 5  * @site www.qqzsh.top
 6  * @company wlgzs
 7  * @create 2019-05-27 8:52
 8  * @description 自定義線程池
 9  */
10 public class Constans {
11 
12     public static final int MAX_THREAD_COUNT = getSystemProcessCount();
13     private static final int MAX_IMUMPOOLSIZE = MAX_THREAD_COUNT;
14 
15     /**
16      * 自定義線程池
17      */
18     private static ExecutorService MY_THREAD_POOL;
19     /**
20      * 自定義線程池
21      */
22     public static ExecutorService getMyThreadPool(){
23         if(MY_THREAD_POOL == null){
24             MY_THREAD_POOL = Executors.newFixedThreadPool(MAX_IMUMPOOLSIZE);
25         }
26         return MY_THREAD_POOL;
27     }
28 
29     /**
30      * 線程池
31      */
32     private static ThreadPoolExecutor threadPool;
33 
34     /**
35      * 單例，單任務 線程池
36      * @return
37      */
38     public static ThreadPoolExecutor getThreadPool(){
39         if(threadPool == null){
40             threadPool = new ThreadPoolExecutor(MAX_IMUMPOOLSIZE, MAX_IMUMPOOLSIZE, 3, TimeUnit.SECONDS,
41                     new ArrayBlockingQueue<>(16),
42                     new ThreadPoolExecutor.CallerRunsPolicy()
43             );
44         }
45         return threadPool;
46     }
47 
48     /**
49      * 獲取服務器cpu核數
50      * @return
51      */
52     private static int getSystemProcessCount(){
53         return Runtime.getRuntime().availableProcessors();
54     }
55 }

客戶端多線程下載類MultiPartDownLoad

  1 import java.io.File;
  2 import java.io.IOException;
  3 import java.io.InputStream;
  4 import java.io.RandomAccessFile;
  5 import java.net.HttpURLConnection;
  6 import java.net.URL;
  7 import java.util.UUID;
  8 import java.util.concurrent.CountDownLatch;
  9 import java.util.concurrent.ExecutorService;
 10 import java.util.concurrent.locks.ReentrantLock;
 11 
 12 /**
 13  * @author zsh
 14  * @site www.qqzsh.top
 15  * @company wlgzs
 16  * @create 2019-05-27 8:53
 17  * @description 多線程下載主程序
 18  */
 19 public class MultiPartDownLoad {
 20     /**
 21      * 線程下載成功標志
 22      */
 23     private static int flag = 0;
 24 
 25     /**
 26      * 服務器請求路徑
 27      */
 28     private String serverPath;
 29     /**
 30      * 本地路徑
 31      */
 32     private String localPath;
 33     /**
 34      * 線程計數同步輔助
 35      */
 36     private CountDownLatch latch;
 37     /**
 38      * 定長線程池
 39      */
 40     private static ExecutorService threadPool;
 41 
 42     public MultiPartDownLoad(String serverPath, String localPath) {
 43         this.serverPath = serverPath;
 44         this.localPath = localPath;
 45     }
 46 
 47     public boolean executeDownLoad() {
 48         try {
 49             URL url = new URL(serverPath);
 50             HttpURLConnection conn = (HttpURLConnection) url.openConnection();
 51             //設置超時時間
 52             conn.setConnectTimeout(5000);
 53             //設置請求方式
 54             conn.setRequestMethod("GET");
 55             conn.setRequestProperty("Connection", "Keep-Alive");
 56             int code = conn.getResponseCode();
 57             if (code != 200) {
 58                 System.out.println(String.format("無效網絡地址：%s", serverPath));
 59                 return false;
 60             }
 61             //服務器返回的數據的長度，實際上就是文件的長度,單位是字節
 62             // int length = conn.getContentLength();  //文件超過2G會有問題
 63             long length = getRemoteFileSize(serverPath);
 64 
 65             System.out.println("遠程文件總長度:" + length + "字節(B)，"+length/Math.pow(2,20)+"MB");
 66             RandomAccessFile raf = new RandomAccessFile(localPath, "rwd");
 67             //指定創建的文件的長度
 68             raf.setLength(length);
 69             raf.close();
 70             //分割文件
 71             int partCount = Constans.MAX_THREAD_COUNT;
 72             int partSize = (int)(length / partCount);
 73             latch = new CountDownLatch(partCount);
 74             threadPool = Constans.getMyThreadPool();
 75             for (int threadId = 1; threadId <= partCount; threadId++) {
 76                 // 每一個線程下載的開始位置
 77                 long startIndex = (threadId - 1) * partSize;
 78                 // 每一個線程下載的開始位置
 79                 long endIndex = startIndex + partSize - 1;
 80                 if (threadId == partCount) {
 81                     //最后一個線程下載的長度稍微長一點
 82                     endIndex = length;
 83                 }
 84                 System.out.println("線程" + threadId + "下載:" + startIndex + "字節~" + endIndex + "字節");
 85                 threadPool.execute(new DownLoadThread(threadId, startIndex, endIndex, latch));
 86             }
 87             latch.await();
 88             if(flag == 0){
 89                 return true;
 90             }
 91         } catch (Exception e) {
 92             System.out.println(String.format("文件下載失敗，文件地址：%s,失敗原因：%s", serverPath, e.getMessage()));
 93         }
 94         return false;
 95     }
 96 
 97     /**
 98      * 內部類用於實現下載
 99      */
100     public class DownLoadThread implements Runnable {
101 
102         /**
103          * 線程ID
104          */
105         private int threadId;
106         /**
107          * 下載起始位置
108          */
109         private long startIndex;
110         /**
111          * 下載結束位置
112          */
113         private long endIndex;
114 
115         private CountDownLatch latch;
116 
117         DownLoadThread(int threadId, long startIndex, long endIndex, CountDownLatch latch) {
118             this.threadId = threadId;
119             this.startIndex = startIndex;
120             this.endIndex = endIndex;
121             this.latch = latch;
122         }
123 
124         @Override
125         public void run() {
126             try {
127                 System.out.println("線程" + threadId + "正在下載...");
128                 URL url = new URL(serverPath);
129                 HttpURLConnection conn = (HttpURLConnection) url.openConnection();
130                 conn.setRequestProperty("Connection", "Keep-Alive");
131                 conn.setRequestMethod("GET");
132                 //請求服務器下載部分的文件的指定位置
133                 conn.setRequestProperty("Range", "bytes=" + startIndex + "-" + endIndex);
134                 conn.setConnectTimeout(5000);
135                 int code = conn.getResponseCode();
136                 System.out.println("線程" + threadId + "請求返回code=" + code);
137                 //返回資源
138                 InputStream is = conn.getInputStream();
139                 RandomAccessFile raf = new RandomAccessFile(localPath, "rwd");
140                 //隨機寫文件的時候從哪個位置開始寫
141                 //定位文件
142                 raf.seek(startIndex);
143                 int len;
144                 byte[] buffer = new byte[1024];
145                 int realLen = 0;
146                 while ((len = is.read(buffer)) != -1) {
147                     realLen += len;
148                     raf.write(buffer, 0, len);
149                 }
150                 System.out.println("線程" + threadId + "下載文件大小=" + realLen/Math.pow(2,20)+"MB");
151                 is.close();
152                 raf.close();
153                 System.out.println("線程" + threadId + "下載完畢");
154             } catch (Exception e) {
155                 //線程下載出錯
156                 MultiPartDownLoad.flag = 1;
157                 System.out.println(e.getMessage());
158             } finally {
159                 //計數值減一
160                 latch.countDown();
161             }
162         }
163     }
164 
165     /**
166      * 內部方法，獲取遠程文件大小
167      * @param remoteFileUrl
168      * @return
169      * @throws IOException
170      */
171     private long getRemoteFileSize(String remoteFileUrl) throws IOException {
172         long fileSize;
173         HttpURLConnection httpConnection = (HttpURLConnection) new URL(remoteFileUrl).openConnection();
174         httpConnection.setRequestMethod("HEAD");
175         int responseCode = 0;
176         try {
177             responseCode = httpConnection.getResponseCode();
178         } catch (IOException e) {
179             e.printStackTrace();
180         }
181         if (responseCode >= 400) {
182             System.out.println("Web服務器響應錯誤!請稍后重試");
183             return 0;
184         }
185         String sHeader;
186         for (int i = 1;; i++) {
187             sHeader = httpConnection.getHeaderFieldKey(i);
188             if ("Content-Length".equals(sHeader)) {
189                 fileSize = Long.parseLong(httpConnection.getHeaderField(sHeader));
190                 break;
191             }
192         }
193         return fileSize;
194     }
195 
196     /**
197      * 下載文件執行器
198      * @param serverPath
199      * @return
200      */
201     public synchronized static String downLoad(String serverPath) {
202         ReentrantLock lock = new ReentrantLock();
203         lock.lock();
204 
205         String[] names = serverPath.split("\\.");
206         if (names.length <= 0) {
207             return null;
208         }
209         String fileTypeName = names[names.length - 1];
210         String localPath = String.format("%s.%s", new File("").getAbsolutePath()+"\\"+UUID.randomUUID(),fileTypeName);
211         MultiPartDownLoad m = new MultiPartDownLoad(serverPath, localPath);
212         long startTime = System.currentTimeMillis();
213         boolean flag = false;
214         try{
215             flag = m.executeDownLoad();
216             long endTime = System.currentTimeMillis();
217             if(flag){
218                 System.out.println("文件下載結束,共耗時" + (endTime - startTime)+ "ms");
219                 return localPath;
220             }
221             System.out.println("文件下載失敗");
222             return null;
223         }catch (Exception ex){
224             System.out.println(ex.getMessage());
225             return null;
226         }finally {
227             // 重置 下載狀態
228             MultiPartDownLoad.flag = 0;
229             if(!flag){
230                 File file = new File(localPath);
231                 file.delete();
232             }
233             lock.unlock();
234         }
235     }
236 }