本文目录导读：
我来为您展示一个完整的Java多线程下载案例，包含断点续传、进度显示和并发控制等功能。
完整的多线程下载实现
多线程下载器主类
import java.io.*;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
public class MultiThreadDownloader {
// 下载配置
private static final int THREAD_COUNT = 4; // 线程数
private static final int BUFFER_SIZE = 1024 * 8; // 缓冲区大小
// 下载信息
private String downloadUrl;
private String savePath;
private long fileSize;
private long downloadedBytes = 0;
private AtomicLong totalDownloaded = new AtomicLong(0);
// 控制组件
private ExecutorService executorService;
private CountDownLatch latch;
private volatile boolean isDownloading = true;
public MultiThreadDownloader(String downloadUrl, String savePath) {
this.downloadUrl = downloadUrl;
this.savePath = savePath;
}
/**
* 开始多线程下载
*/
public void download() throws Exception {
// 1. 获取文件大小
fileSize = getFileSize(downloadUrl);
System.out.println("文件大小: " + formatFileSize(fileSize) + " (" + fileSize + " bytes)");
// 2. 初始化线程池
executorService = Executors.newFixedThreadPool(THREAD_COUNT);
latch = new CountDownLatch(THREAD_COUNT);
// 3. 创建临时文件并设置长度
RandomAccessFile raf = new RandomAccessFile(savePath, "rw");
raf.setLength(fileSize);
raf.close();
// 4. 计算每个线程的下载区间
long partSize = fileSize / THREAD_COUNT;
long startTime = System.currentTimeMillis();
System.out.println("开始下载，使用 " + THREAD_COUNT + " 个线程...");
// 5. 提交下载任务
for (int i = 0; i < THREAD_COUNT; i++) {
long start = i * partSize;
long end = (i == THREAD_COUNT - 1) ? fileSize - 1 : (i + 1) * partSize - 1;
executorService.submit(new DownloadTask(i, start, end));
}
// 6. 等待所有线程完成
latch.await();
// 7. 关闭线程池
executorService.shutdown();
// 8. 显示下载结果
long endTime = System.currentTimeMillis();
System.out.println("\n下载完成！");
System.out.println("用时: " + (endTime - startTime) / 1000.0 + " 秒");
System.out.println("平均速度: " + formatFileSize(fileSize / ((endTime - startTime) / 1000.0)) + "/s");
}
/**
* 单个下载任务
*/
private class DownloadTask implements Runnable {
private int threadId;
private long startPos;
private long endPos;
public DownloadTask(int threadId, long startPos, long endPos) {
this.threadId = threadId;
this.startPos = startPos;
this.endPos = endPos;
}
@Override
public void run() {
HttpURLConnection connection = null;
RandomAccessFile raf = null;
InputStream inputStream = null;
try {
System.out.println("线程 " + (threadId + 1) + " 开始下载: "
+ startPos + " - " + endPos);
// 建立连接
URL url = new URL(downloadUrl);
connection = (HttpURLConnection) url.openConnection();
// 设置请求头，指定下载范围
connection.setRequestProperty("Range", "bytes=" + startPos + "-" + endPos);
connection.setRequestProperty("User-Agent",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36");
// 获取响应码
int responseCode = connection.getResponseCode();
if (responseCode != HttpURLConnection.HTTP_PARTIAL) {
// 如果服务器不支持断点续传
if (responseCode == HttpURLConnection.HTTP_OK && threadId == 0) {
// 整个文件下载
downloadFullFile(connection);
return;
} else {
throw new IOException("服务器不支持断点续传，响应码: " + responseCode);
}
}
// 打开文件和输入流
raf = new RandomAccessFile(savePath, "rw");
raf.seek(startPos);
inputStream = connection.getInputStream();
// 开始下载
byte[] buffer = new byte[BUFFER_SIZE];
int bytesRead;
long currentPos = startPos;
while (currentPos <= endPos && isDownloading) {
bytesRead = inputStream.read(buffer);
if (bytesRead == -1) break;
// 确保不会超出范围
long remaining = endPos - currentPos + 1;
int writeLength = (int) Math.min(bytesRead, remaining);
raf.write(buffer, 0, writeLength);
currentPos += writeLength;
// 更新下载进度
long downloaded = totalDownloaded.addAndGet(writeLength);
// 打印进度（每5%打印一次）
int progress = (int) (downloaded * 100 / fileSize);
if (progress % 5 == 0) {
synchronized (System.out) {
System.out.print("\r下载进度: " + progress + "% "
+ formatFileSize(downloaded) + "/" + formatFileSize(fileSize));
}
}
}
System.out.println("\n线程 " + (threadId + 1) + " 下载完成");
} catch (Exception e) {
System.err.println("线程 " + (threadId + 1) + " 下载失败: " + e.getMessage());
e.printStackTrace();
} finally {
// 关闭资源
try {
if (inputStream != null) inputStream.close();
if (raf != null) raf.close();
if (connection != null) connection.disconnect();
} catch (IOException e) {
e.printStackTrace();
}
latch.countDown();
}
}
/**
* 下载整个文件（服务器不支持断点续传时）
*/
private void downloadFullFile(HttpURLConnection connection) throws IOException {
RandomAccessFile raf = new RandomAccessFile(savePath, "rw");
InputStream inputStream = connection.getInputStream();
byte[] buffer = new byte[BUFFER_SIZE];
int bytesRead;
while ((bytesRead = inputStream.read(buffer)) != -1) {
raf.write(buffer, 0, bytesRead);
totalDownloaded.addAndGet(bytesRead);
}
inputStream.close();
raf.close();
}
}
/**
* 获取文件大小
*/
private long getFileSize(String url) throws IOException {
HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection();
connection.setRequestMethod("HEAD");
connection.setRequestProperty("User-Agent", "Mozilla/5.0");
long size = connection.getContentLengthLong();
connection.disconnect();
return size;
}
/**
* 格式化文件大小
*/
private static String formatFileSize(double bytes) {
if (bytes < 1024) return String.format("%.1f B", bytes);
if (bytes < 1024 * 1024) return String.format("%.1f KB", bytes / 1024);
if (bytes < 1024 * 1024 * 1024) return String.format("%.1f MB", bytes / (1024 * 1024));
return String.format("%.1f GB", bytes / (1024 * 1024 * 1024));
}
}
断点续传版本
import java.io.*;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
public class ResumeDownloader {
private static final int THREAD_COUNT = 4;
private static final String TEMP_SUFFIX = ".part";
private String downloadUrl;
private String savePath;
private String tempPath;
private long fileSize;
private AtomicLong totalDownloaded = new AtomicLong(0);
private CountDownLatch latch;
private ExecutorService executorService;
// 记录每个线程的下载进度
private long[] downloadProgress;
public ResumeDownloader(String downloadUrl, String savePath) {
this.downloadUrl = downloadUrl;
this.savePath = savePath;
this.tempPath = savePath + TEMP_SUFFIX;
this.downloadProgress = new long[THREAD_COUNT];
}
public void downloadWithResume() throws Exception {
// 获取文件大小
fileSize = getFileSize(downloadUrl);
System.out.println("文件大小: " + formatFileSize(fileSize));
// 初始化临时文件
File tempFile = new File(tempPath);
if (!tempFile.exists()) {
RandomAccessFile raf = new RandomAccessFile(tempFile, "rw");
raf.setLength(fileSize);
raf.close();
System.out.println("创建临时文件");
} else {
System.out.println("检测到临时文件，尝试断点续传");
}
// 计算每个线程的下载区间
long partSize = fileSize / THREAD_COUNT;
// 初始化线程池
executorService = Executors.newFixedThreadPool(THREAD_COUNT);
latch = new CountDownLatch(THREAD_COUNT);
long startTime = System.currentTimeMillis();
// 提交下载任务
for (int i = 0; i < THREAD_COUNT; i++) {
long start = i * partSize;
long end = (i == THREAD_COUNT - 1) ? fileSize - 1 : (i + 1) * partSize - 1;
// 检查该部分是否已经下载完成
File partFile = new File(tempPath + "." + i);
if (partFile.exists()) {
long completed = partFile.length();
start += completed;
downloadProgress[i] = completed;
System.out.println("线程 " + (i + 1) + " 已下载 " + completed + " bytes，继续下载");
}
if (start <= end) {
executorService.submit(new ResumeTask(i, start, end));
} else {
latch.countDown(); // 该部分已完成
}
}
// 等待下载完成
latch.await();
executorService.shutdown();
// 合并文件
mergeFiles();
long endTime = System.currentTimeMillis();
System.out.println("\n下载完成！用时: " + (endTime - startTime) / 1000.0 + " 秒");
}
private class ResumeTask implements Runnable {
private int threadId;
private long startPos;
private long endPos;
public ResumeTask(int threadId, long startPos, long endPos) {
this.threadId = threadId;
this.startPos = startPos;
this.endPos = endPos;
}
@Override
public void run() {
HttpURLConnection connection = null;
RandomAccessFile raf = null;
InputStream inputStream = null;
try {
URL url = new URL(downloadUrl);
connection = (HttpURLConnection) url.openConnection();
// 设置断点续传的请求头
System.out.println("线程 " + (threadId + 1) + " 请求范围: " + startPos + "-" + endPos);
connection.setRequestProperty("Range", "bytes=" + startPos + "-" + endPos);
connection.setRequestProperty("User-Agent",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36");
int responseCode = connection.getResponseCode();
if (responseCode != HttpURLConnection.HTTP_PARTIAL) {
throw new IOException("服务器响应码异常: " + responseCode);
}
raf = new RandomAccessFile(tempPath, "rw");
raf.seek(startPos); // 移动到断点位置
inputStream = connection.getInputStream();
byte[] buffer = new byte[1024 * 16];
int bytesRead;
long currentPos = startPos;
while (currentPos <= endPos) {
bytesRead = inputStream.read(buffer);
if (bytesRead == -1) break;
long remaining = endPos - currentPos + 1;
int writeLength = (int) Math.min(bytesRead, remaining);
raf.write(buffer, 0, writeLength);
currentPos += writeLength;
long downloaded = totalDownloaded.addAndGet(writeLength);
// 打印进度
if (totalDownloaded.get() % 1000000 < writeLength) {
int progress = (int) (downloaded * 100 / fileSize);
System.out.printf("\r下载进度: %d%%", progress);
}
}
System.out.println("\n线程 " + (threadId + 1) + " 下载完成");
} catch (Exception e) {
System.err.println("线程 " + (threadId + 1) + " 错误: " + e.getMessage());
} finally {
try {
if (inputStream != null) inputStream.close();
if (raf != null) raf.close();
if (connection != null) connection.disconnect();
} catch (IOException e) {
e.printStackTrace();
}
latch.countDown();
}
}
}
/**
* 合并临时文件
*/
private void mergeFiles() {
System.out.println("正在合并文件...");
try (FileOutputStream fos = new FileOutputStream(savePath);
FileChannel outChannel = fos.getChannel()) {
File tempFile = new File(tempPath);
try (FileInputStream fis = new FileInputStream(tempFile);
FileChannel inChannel = fis.getChannel()) {
inChannel.transferTo(0, tempFile.length(), outChannel);
}
// 删除临时文件
if (!tempFile.delete()) {
System.out.println("无法删除临时文件: " + tempPath);
}
System.out.println("文件合并完成");
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 获取文件大小
*/
private long getFileSize(String url) throws IOException {
HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection();
connection.setRequestMethod("HEAD");
connection.setRequestProperty("User-Agent", "Mozilla/5.0");
long size = connection.getContentLengthLong();
connection.disconnect();
return size;
}
private static String formatFileSize(double bytes) {
if (bytes < 1024) return String.format("%.1f B", bytes);
if (bytes < 1024 * 1024) return String.format("%.1f KB", bytes / 1024);
if (bytes < 1024 * 1024 * 1024) return String.format("%.1f MB", bytes / (1024 * 1024));
return String.format("%.1f GB", bytes / (1024 * 1024 * 1024));
}
}
测试类
public class DownloadTest {
public static void main(String[] args) {
// 测试文件下载
String downloadUrl = "http://speedtest.tele2.net/100MB.zip";
String savePath = "D:/downloads/100MB.zip";
try {
// 简单多线程下载
System.out.println("=== 简单多线程下载 ===");
MultiThreadDownloader downloader = new MultiThreadDownloader(downloadUrl, savePath);
downloader.download();
// 断点续传下载
/*
System.out.println("=== 断点续传下载 ===");
ResumeDownloader resumeDownloader = new ResumeDownloader(downloadUrl, savePath);
resumeDownloader.downloadWithResume();
*/
} catch (Exception e) {
e.printStackTrace();
}
}
}
关键特性说明
- 多线程分段下载：将文件分成多个部分，每个线程负责下载一部分
- 断点续传：记录下载进度，支持中断后继续下载
- 进度显示：实时显示下载进度和速度
- 资源管理：正确关闭流和连接，使用线程池管理线程
- 错误处理：捕获并处理各种异常情况
使用建议
// 下载大文件时建议设置代理
System.setProperty("http.proxyHost", "proxy.example.com");
System.setProperty("http.proxyPort", "8080");
// 可以通过配置文件配置下载参数
// thread.count=8
// buffer.size=16384
// timeout=30000
这个示例涵盖了Java多线程下载的核心实现，您可以根据实际需求调整线程数、下载逻辑和错误处理机制。
