diff --git a/Aria/src/main/java/com/arialyy/aria/core/common/AbsFileer.java b/Aria/src/main/java/com/arialyy/aria/core/common/AbsFileer.java index 91fbc955..cae8612f 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/common/AbsFileer.java +++ b/Aria/src/main/java/com/arialyy/aria/core/common/AbsFileer.java @@ -22,6 +22,7 @@ import com.arialyy.aria.core.download.DownloadTaskEntity; import com.arialyy.aria.core.inf.AbsNormalEntity; import com.arialyy.aria.core.inf.AbsTaskEntity; import com.arialyy.aria.core.inf.IEventListener; +import com.arialyy.aria.core.manager.ThreadTaskManager; import com.arialyy.aria.orm.DbEntity; import com.arialyy.aria.util.ALog; import com.arialyy.aria.util.CommonUtil; @@ -36,8 +37,6 @@ import java.util.Properties; import java.util.Set; import java.util.Timer; import java.util.TimerTask; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; /** * Created by AriaL on 2017/7/1. @@ -63,7 +62,6 @@ public abstract class AbsFileer keys = mTask.keySet(); - //for (Integer key : keys) { - // AbsThreadTask task = mTask.get(key); - // if (task != null && !task.isThreadComplete()) { - // - // task.getConfig().START_LOCATION = task.getConfig().THREAD_RECORD.startLocation; - // mConstance.resetState(); - // startTimer(); - // ALog.d(TAG, String.format("任务【%s】开始重试,线程__%s__【开始位置:%s,结束位置:%s】", mEntity.getFileName(), - // key, task.getConfig().START_LOCATION, task.getConfig().END_LOCATION)); - // if (!mFixedThreadPool.isShutdown()) { - // mFixedThreadPool.execute(task); - // } else { - // ALog.w(TAG, "线程池已关闭"); - // mListener.onFail(true); - // } - // } - //} } /** @@ -707,8 +669,7 @@ public abstract class AbsFileer> - implements Runnable { + implements Callable { /** * 线程重试次数 */ @@ -352,7 +354,8 @@ public abstract class AbsThreadTask> mThreadTasks = new ConcurrentHashMap<>(); - - private ThreadTaskManager() { - - } - - public static ThreadTaskManager getInstance() { - if (INSTANCE == null) { - synchronized (ThreadTaskManager.class) { - INSTANCE = new ThreadTaskManager(); - } - } - return INSTANCE; - } - - /** - * 添加单条线程记录 - * - * @param key 任务对应的key - * @param threadTask 线程任务 - */ - public void addTask(String key, AbsThreadTask threadTask) { - if (mThreadTasks.get(key) == null) { - mThreadTasks.put(key, new ArrayList()); - } - mThreadTasks.get(key).add(threadTask); - } - - /** - * 删除对应的任务的线程记录 - * - * @param key 任务对应的key - */ - public void removeTask(String key) { - for (Iterator>> iter = mThreadTasks.entrySet().iterator(); - iter.hasNext(); ) { - Map.Entry> entry = iter.next(); - if (key.equals(entry.getKey())) { - List list = mThreadTasks.get(key); - if (list != null && !list.isEmpty()) { - list.clear(); - } - iter.remove(); - } - } - } - - - -} diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/downloader/FtpThreadTask.java b/Aria/src/main/java/com/arialyy/aria/core/download/downloader/FtpThreadTask.java index 314a7230..bbaa67ed 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/downloader/FtpThreadTask.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/downloader/FtpThreadTask.java @@ -53,10 +53,10 @@ class FtpThreadTask extends AbsFtpThreadTask isBlock = STATE.TASK_RECORD.isBlock; } - @Override public void run() { + @Override public DownloadTaskEntity call() throws Exception { if (mConfig.THREAD_RECORD.isComplete) { handleComplete(); - return; + return mTaskEntity; } mChildCurrentLocation = mConfig.START_LOCATION; FTPClient client = null; @@ -69,7 +69,7 @@ class FtpThreadTask extends AbsFtpThreadTask client = createClient(); if (client == null) { fail(mChildCurrentLocation, "ftp client 创建失败", null); - return; + return mTaskEntity; } if (mConfig.START_LOCATION > 0) { client.setRestartOffset(mConfig.START_LOCATION); @@ -81,7 +81,7 @@ class FtpThreadTask extends AbsFtpThreadTask String.format("获取文件信息错误,错误码为:%s,msg:%s", reply, client.getReplyString()), null); client.disconnect(); - return; + return mTaskEntity; } String remotePath = new String(mTaskEntity.getUrlEntity().remotePath.getBytes(charSet), SERVER_CHARSET); @@ -93,7 +93,7 @@ class FtpThreadTask extends AbsFtpThreadTask String.format("获取流失败,错误码为:%s,msg:%s", reply, client.getReplyString()), null); client.disconnect(); - return; + return mTaskEntity; } if (isOpenDynamicFile) { @@ -118,6 +118,7 @@ class FtpThreadTask extends AbsFtpThreadTask e.printStackTrace(); } } + return mTaskEntity; } /** @@ -164,7 +165,7 @@ class FtpThreadTask extends AbsFtpThreadTask foc = fos.getChannel(); fic = Channels.newChannel(is); ByteBuffer bf = ByteBuffer.allocate(mBufSize); - while ((len = fic.read(bf)) != -1) { + while (!Thread.currentThread().isInterrupted() && (len = fic.read(bf)) != -1) { if (isBreak()) { break; } @@ -215,7 +216,7 @@ class FtpThreadTask extends AbsFtpThreadTask file.seek(mConfig.START_LOCATION); byte[] buffer = new byte[mBufSize]; int len; - while ((len = is.read(buffer)) != -1) { + while (!Thread.currentThread().isInterrupted() && (len = is.read(buffer)) != -1) { if (isBreak()) { break; } diff --git a/Aria/src/main/java/com/arialyy/aria/core/download/downloader/HttpThreadTask.java b/Aria/src/main/java/com/arialyy/aria/core/download/downloader/HttpThreadTask.java index 9e75a3f4..9bc2ec5b 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/download/downloader/HttpThreadTask.java +++ b/Aria/src/main/java/com/arialyy/aria/core/download/downloader/HttpThreadTask.java @@ -33,6 +33,7 @@ import java.net.MalformedURLException; import java.net.URL; import java.nio.ByteBuffer; import java.nio.channels.Channels; +import java.nio.channels.ClosedByInterruptException; import java.nio.channels.FileChannel; import java.nio.channels.ReadableByteChannel; @@ -56,10 +57,10 @@ final class HttpThreadTask extends AbsThreadTask> mThreadTasks = new HashMap<>(); + + public static ThreadTaskManager getInstance() { + if (INSTANCE == null) { + synchronized (LOCK) { + INSTANCE = new ThreadTaskManager(); + } + } + return INSTANCE; + } + + private ThreadTaskManager() { + mExePool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); + } + + /** + * 启动线程任务 + * + * @param key 任务对应的key{@link AbsTaskEntity#getKey()} + * @param threadTask 线程任务{@link AbsThreadTask} + */ + public void startThread(String key, AbsThreadTask threadTask) { + if (mExePool.isShutdown()) { + ALog.e(TAG, "线程池已经关闭"); + return; + } + key = getKey(key); + Set temp = mThreadTasks.get(key); + if (temp == null) { + temp = new HashSet<>(); + mThreadTasks.put(key, temp); + } + temp.add(mExePool.submit(threadTask)); + } + + /** + * 停止任务的所有线程 + * + * @param key 任务对应的key{@link AbsTaskEntity#getKey()} + */ + public void stopTaskThread(String key) { + if (mExePool.isShutdown()) { + ALog.e(TAG, "线程池已经关闭"); + return; + } + Set temp = mThreadTasks.get(getKey(key)); + for (Future future : temp) { + if (future.isDone() || future.isCancelled()) { + continue; + } + future.cancel(true); + } + mThreadTasks.remove(key); + } + + /** + * 重试线程任务 + * + * @param task 线程任务 + */ + public void retryThread(AbsThreadTask task) { + if (mExePool.isShutdown()) { + ALog.e(TAG, "线程池已经关闭"); + return; + } + mExePool.submit(task); + } + + /** + * map中的key + * + * @param key 任务的key{@link AbsTaskEntity#getKey()} + * @return 转换后的map中的key + */ + private String getKey(String key) { + return CommonUtil.getStrMd5(key); + } } diff --git a/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/FtpThreadTask.java b/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/FtpThreadTask.java index bfb59627..c3a233f6 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/FtpThreadTask.java +++ b/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/FtpThreadTask.java @@ -50,7 +50,7 @@ class FtpThreadTask extends AbsFtpThreadTask { return mAridManager.getUploadConfig().getMaxSpeed(); } - @Override public void run() { + @Override public UploadTaskEntity call() throws Exception { //当前子线程的下载位置 mChildCurrentLocation = mConfig.START_LOCATION; FTPClient client = null; @@ -60,7 +60,9 @@ class FtpThreadTask extends AbsFtpThreadTask { String.format("任务【%s】线程__%s__开始上传【开始位置 : %s,结束位置:%s】", mConfig.TEMP_FILE.getName(), mConfig.THREAD_ID, mConfig.START_LOCATION, mConfig.END_LOCATION)); client = createClient(); - if (client == null) return; + if (client == null) { + return mTaskEntity; + } initPath(); client.makeDirectory(dir); client.changeWorkingDirectory(dir); @@ -69,7 +71,7 @@ class FtpThreadTask extends AbsFtpThreadTask { if (!FTPReply.isPositivePreliminary(reply) && reply != FTPReply.FILE_ACTION_OK) { fail(mChildCurrentLocation, "上传失败,错误码为:" + reply + ",msg:" + client.getReplyString(), null); client.disconnect(); - return; + return mTaskEntity; } file = new BufferedRandomAccessFile(mConfig.TEMP_FILE, "rwd", mBufSize); @@ -79,7 +81,7 @@ class FtpThreadTask extends AbsFtpThreadTask { } upload(client, file); if (isBreak()) { - return; + return mTaskEntity; } ALog.i(TAG, String.format("任务【%s】线程__%s__上传完毕", mConfig.TEMP_FILE.getName(), mConfig.THREAD_ID)); @@ -110,6 +112,7 @@ class FtpThreadTask extends AbsFtpThreadTask { e.printStackTrace(); } } + return mTaskEntity; } private void initPath() throws UnsupportedEncodingException { diff --git a/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/HttpThreadTask.java b/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/HttpThreadTask.java index 822ba74e..1978b9ed 100644 --- a/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/HttpThreadTask.java +++ b/Aria/src/main/java/com/arialyy/aria/core/upload/uploader/HttpThreadTask.java @@ -57,12 +57,12 @@ class HttpThreadTask extends AbsThreadTask { isNotNetRetry = mAridManager.getUploadConfig().isNotNetRetry(); } - @Override public void run() { + @Override public UploadTaskEntity call() throws Exception { File uploadFile = new File(mEntity.getFilePath()); if (!uploadFile.exists()) { ALog.e(TAG, String.format("【%s】,文件不存在。", mEntity.getFilePath())); fail(); - return; + return mTaskEntity; } mListener.onPre(); URL url; @@ -103,6 +103,7 @@ class HttpThreadTask extends AbsThreadTask { e.printStackTrace(); fail(); } + return mTaskEntity; } private void fail() { @@ -199,7 +200,7 @@ class HttpThreadTask extends AbsThreadTask { if (status == HttpURLConnection.HTTP_OK) { BufferedReader reader = new BufferedReader(new InputStreamReader(mHttpConn.getInputStream())); String line; - while ((line = reader.readLine()) != null) { + while (!Thread.currentThread().isInterrupted() && (line = reader.readLine()) != null) { response.append(line); } reader.close(); diff --git a/app/src/main/java/com/arialyy/simple/download/SingleTaskActivity.java b/app/src/main/java/com/arialyy/simple/download/SingleTaskActivity.java index 0bb2cc84..53c76670 100644 --- a/app/src/main/java/com/arialyy/simple/download/SingleTaskActivity.java +++ b/app/src/main/java/com/arialyy/simple/download/SingleTaskActivity.java @@ -222,9 +222,9 @@ public class SingleTaskActivity extends BaseActivity { startD(); break; case R.id.stop: - //Aria.download(this).load(DOWNLOAD_URL).stop(); - startActivity(new Intent(this, SingleTaskActivity.class)); - Aria.download(this).unRegister(); + Aria.download(this).load(DOWNLOAD_URL).stop(); + //startActivity(new Intent(this, SingleTaskActivity.class)); + //Aria.download(this).unRegister(); //Aria.download(this).load(DOWNLOAD_URL).removeRecord(); break; case R.id.cancel: