FTP/HTTP 多线程断点续传下载
支持FTP和HTTP协议的多线程断点续传下载程序，使用C++实现，支持Windows和Linux平台。
系统架构
graph TD
A[用户界面] --> B[下载管理器]
B --> C[HTTP下载器]
B --> D[FTP下载器]
C --> E[HTTP连接池]
D --> F[FTP连接池]
B --> G[断点续传管理器]
G --> H[下载状态存储]
B --> I[任务调度器]
I --> J[线程池]
核心实现代码
1. 主程序 (main.cpp)
#include <iostream>
#include <string>
#include <vector>
#include <thread>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <fstream>
#include <sstream>
#include <cstdlib>
#include <cstring>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <curl/curl.h>
#include <json/json.h>
#ifdef _WIN32
#include <windows.h>
#include <wininet.h>
#pragma comment(lib, "wininet.lib")
#pragma comment(lib, "ws2_32.lib")
#else
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <fcntl.h>
#include <errno.h>
#endif
using namespace std;
// 下载任务类
class DownloadTask {
public:
string url;
string savePath;
string fileName;
long fileSize;
long downloadedSize;
int threadCount;
vector<long> partSizes;
vector<long> partDownloaded;
atomic<bool> isPaused;
atomic<bool> isCancelled;
atomic<int> activeThreads;
mutex mtx;
condition_variable cv;
DownloadTask(const string& u, const string& path, int threads = 4)
: url(u), savePath(path), threadCount(threads),
fileSize(0), downloadedSize(0),
isPaused(false), isCancelled(false), activeThreads(0) {}
void parseUrl() {
// 解析URL获取文件名
size_t pos = url.find_last_of("/");
if (pos != string::npos) {
fileName = url.substr(pos + 1);
if (fileName.empty()) fileName = "index.html";
} else {
fileName = "downloaded_file";
}
// 如果保存路径是目录，则使用URL中的文件名
if (savePath.back() == '/' || savePath.back() == '\\') {
savePath += fileName;
}
}
void calculateParts() {
partSizes.resize(threadCount);
partDownloaded.resize(threadCount, 0);
long partSize = fileSize / threadCount;
for (int i = 0; i < threadCount; i++) {
partSizes[i] = partSize;
}
partSizes[threadCount - 1] += fileSize % threadCount;
}
};
// 下载管理器
class DownloadManager {
public:
DownloadManager(int maxThreads = 8)
: maxThreads(maxThreads), running(true) {
threadPool.reserve(maxThreads);
for (int i = 0; i < maxThreads; i++) {
threadPool.emplace_back(&DownloadManager::workerThread, this);
}
}
~DownloadManager() {
running = false;
cv.notify_all();
for (auto& t : threadPool) {
if (t.joinable()) t.join();
}
}
void addTask(DownloadTask* task) {
unique_lock<mutex> lock(mtx);
tasks.push_back(task);
lock.unlock();
cv.notify_one();
}
void pauseTask(DownloadTask* task) {
task->isPaused = true;
}
void resumeTask(DownloadTask* task) {
task->isPaused = false;
task->cv.notify_all();
}
void cancelTask(DownloadTask* task) {
task->isCancelled = true;
task->cv.notify_all();
}
void waitForTask(DownloadTask* task) {
unique_lock<mutex> lock(task->mtx);
while (task->activeThreads > 0) {
task->cv.wait(lock);
}
}
private:
vector<thread> threadPool;
vector<DownloadTask*> tasks;
mutex mtx;
condition_variable cv;
int maxThreads;
atomic<bool> running;
void workerThread() {
while (running) {
DownloadTask* task = nullptr;
{
unique_lock<mutex> lock(mtx);
cv.wait(lock, [this] {
return !tasks.empty() || !running;
});
if (!running) break;
if (!tasks.empty()) {
task = tasks.back();
tasks.pop_back();
}
}
if (task) {
processTask(task);
}
}
}
void processTask(DownloadTask* task) {
// 创建目录
string dir = task->savePath.substr(0, task->savePath.find_last_of("/\\"));
#ifdef _WIN32
CreateDirectoryA(dir.c_str(), NULL);
#else
mkdir(dir.c_str(), 0755);
#endif
// 获取文件大小
if (!getFileSize(task)) {
cerr << "Failed to get file size: " << task->url << endl;
return;
}
// 计算分块
task->calculateParts();
// 创建临时文件
ofstream outFile(task->savePath + ".part", ios::binary | ios::out);
outFile.close();
// 设置文件大小
#ifdef _WIN32
HANDLE hFile = CreateFileA((task->savePath + ".part").c_str(),
GENERIC_WRITE, 0, NULL,
CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
if (hFile != INVALID_HANDLE_VALUE) {
LARGE_INTEGER li;
li.QuadPart = task->fileSize;
SetFilePointerEx(hFile, li, NULL, FILE_BEGIN);
SetEndOfFile(hFile);
CloseHandle(hFile);
}
#else
int fd = open((task->savePath + ".part").c_str(), O_WRONLY | O_CREAT, 0644);
if (fd != -1) {
ftruncate(fd, task->fileSize);
close(fd);
}
#endif
// 创建下载线程
vector<thread> downloadThreads;
for (int i = 0; i < task->threadCount; i++) {
task->activeThreads++;
downloadThreads.emplace_back([this, task, i] {
downloadPart(task, i);
});
}
// 等待所有线程完成
for (auto& t : downloadThreads) {
t.join();
}
// 合并文件
if (task->downloadedSize == task->fileSize) {
rename((task->savePath + ".part").c_str(), task->savePath.c_str());
cout << "Download completed: " << task->savePath << endl;
} else {
cout << "Download incomplete: " << task->downloadedSize << "/"
<< task->fileSize << " bytes" << endl;
}
// 清理
delete task;
}
bool getFileSize(DownloadTask* task) {
CURL* curl = curl_easy_init();
if (!curl) return false;
curl_easy_setopt(curl, CURLOPT_URL, task->url.c_str());
curl_easy_setopt(curl, CURLOPT_NOBODY, 1L);
curl_easy_setopt(curl, CURLOPT_HEADER, 1L);
curl_easy_setopt(curl, CURLOPT_FILETIME, 1L);
curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
CURLcode res = curl_easy_perform(curl);
if (res == CURLE_OK) {
double size;
if (curl_easy_getinfo(curl, CURLINFO_CONTENT_LENGTH_DOWNLOAD, &size) == CURLE_OK) {
task->fileSize = static_cast<long>(size);
return true;
}
}
curl_easy_cleanup(curl);
return false;
}
void downloadPart(DownloadTask* task, int partIndex) {
long start = 0, end = 0;
for (int i = 0; i < partIndex; i++) {
start += task->partSizes[i];
}
end = start + task->partSizes[partIndex] - 1;
CURL* curl = curl_easy_init();
if (!curl) {
task->activeThreads--;
return;
}
// 设置断点续传
long localDownloaded = 0;
#ifdef _WIN32
HANDLE hFile = CreateFileA((task->savePath + ".part").c_str(),
GENERIC_READ | GENERIC_WRITE, 0, NULL,
OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, NULL);
if (hFile != INVALID_HANDLE_VALUE) {
LARGE_INTEGER li;
li.QuadPart = start;
SetFilePointerEx(hFile, li, NULL, FILE_BEGIN);
DWORD bytesRead;
char buffer[1024];
while (ReadFile(hFile, buffer, sizeof(buffer), &bytesRead, NULL) && bytesRead > 0) {
localDownloaded += bytesRead;
}
CloseHandle(hFile);
}
#else
int fd = open((task->savePath + ".part").c_str(), O_RDWR);
if (fd != -1) {
lseek(fd, start, SEEK_SET);
char buffer[1024];
ssize_t bytesRead;
while ((bytesRead = read(fd, buffer, sizeof(buffer))) > 0) {
localDownloaded += bytesRead;
}
close(fd);
}
#endif
if (localDownloaded > 0) {
start += localDownloaded;
task->partDownloaded[partIndex] = localDownloaded;
task->downloadedSize += localDownloaded;
}
// 设置Range头
string rangeHeader = "Range: bytes=" + to_string(start) + "-" + to_string(end);
// 打开文件
#ifdef _WIN32
HANDLE hFileOut = CreateFileA((task->savePath + ".part").c_str(),
GENERIC_WRITE, 0, NULL,
OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, NULL);
if (hFileOut == INVALID_HANDLE_VALUE) {
task->activeThreads--;
return;
}
SetFilePointer(hFileOut, start, NULL, FILE_BEGIN);
#else
int fdOut = open((task->savePath + ".part").c_str(), O_WRONLY);
if (fdOut == -1) {
task->activeThreads--;
return;
}
lseek(fdOut, start, SEEK_SET);
#endif
// 设置CURL选项
curl_easy_setopt(curl, CURLOPT_URL, task->url.c_str());
curl_easy_setopt(curl, CURLOPT_RANGE, (to_string(start) + "-" + to_string(end)).c_str());
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writeData);
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &fdOut);
curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0L);
curl_easy_setopt(curl, CURLOPT_PROGRESSFUNCTION, progressCallback);
curl_easy_setopt(curl, CURLOPT_PROGRESSDATA, task);
curl_easy_setopt(curl, CURLOPT_USERAGENT, "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36");
// 执行下载
CURLcode res = curl_easy_perform(curl);
// 清理
if (res == CURLE_OK) {
long httpCode = 0;
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &httpCode);
if (httpCode == 206 || httpCode == 200) {
task->partDownloaded[partIndex] = task->partSizes[partIndex];
task->downloadedSize += task->partSizes[partIndex];
}
}
// 关闭文件
#ifdef _WIN32
CloseHandle(hFileOut);
#else
close(fdOut);
#endif
curl_easy_cleanup(curl);
task->activeThreads--;
task->cv.notify_all();
}
static size_t writeData(void* ptr, size_t size, size_t nmemb, void* userdata) {
int fd = *(int*)userdata;
size_t totalSize = size * nmemb;
#ifdef _WIN32
DWORD bytesWritten;
WriteFile((HANDLE)fd, ptr, totalSize, &bytesWritten, NULL);
return bytesWritten;
#else
return write(fd, ptr, totalSize);
#endif
}
static int progressCallback(void* clientp, curl_off_t dltotal, curl_off_t dlnow,
curl_off_t ultotal, curl_off_t ulnow) {
DownloadTask* task = (DownloadTask*)clientp;
if (task->isCancelled) return 1; // 取消下载
unique_lock<mutex> lock(task->mtx);
while (task->isPaused) {
task->cv.wait(lock);
if (task->isCancelled) return 1;
}
return 0;
}
};
// 用户界面
void showMenu() {
cout << "\n===== 多线程断点续传下载器 =====" << endl;
cout << "1. 添加下载任务" << endl;
cout << "2. 暂停下载任务" << endl;
cout << "3. 继续下载任务" << endl;
cout << "4. 取消下载任务" << endl;
cout << "5. 退出程序" << endl;
cout << "===============================" << endl;
cout << "请选择操作: ";
}
int main() {
// 初始化CURL
curl_global_init(CURL_GLOBAL_ALL);
DownloadManager manager(8);
vector<DownloadTask*> activeTasks;
int choice;
while (true) {
showMenu();
cin >> choice;
switch (choice) {
case 1: {
string url, path;
int threads;
cout << "请输入下载URL: ";
cin >> url;
cout << "请输入保存路径(留空使用当前目录): ";
cin.ignore();
getline(cin, path);
if (path.empty()) path = "./";
cout << "请输入线程数(默认4): ";
string threadInput;
getline(cin, threadInput);
threads = threadInput.empty() ? 4 : stoi(threadInput);
DownloadTask* task = new DownloadTask(url, path, threads);
task->parseUrl();
manager.addTask(task);
activeTasks.push_back(task);
cout << "下载任务已添加: " << url << endl;
break;
}
case 2: {
if (activeTasks.empty()) {
cout << "没有活动的下载任务" << endl;
break;
}
cout << "请选择要暂停的任务(1-" << activeTasks.size() << "): ";
int index;
cin >> index;
if (index > 0 && index <= activeTasks.size()) {
manager.pauseTask(activeTasks[index-1]);
cout << "任务已暂停" << endl;
} else {
cout << "无效的选择" << endl;
}
break;
}
case 3: {
if (activeTasks.empty()) {
cout << "没有活动的下载任务" << endl;
break;
}
cout << "请选择要继续的任务(1-" << activeTasks.size() << "): ";
int index;
cin >> index;
if (index > 0 && index <= activeTasks.size()) {
manager.resumeTask(activeTasks[index-1]);
cout << "任务已继续" << endl;
} else {
cout << "无效的选择" << endl;
}
break;
}
case 4: {
if (activeTasks.empty()) {
cout << "没有活动的下载任务" << endl;
break;
}
cout << "请选择要取消的任务(1-" << activeTasks.size() << "): ";
int index;
cin >> index;
if (index > 0 && index <= activeTasks.size()) {
manager.cancelTask(activeTasks[index-1]);
manager.waitForTask(activeTasks[index-1]);
cout << "任务已取消" << endl;
// 从活动任务列表中移除
activeTasks.erase(activeTasks.begin() + index - 1);
} else {
cout << "无效的选择" << endl;
}
break;
}
case 5:
cout << "正在退出..." << endl;
// 取消所有任务
for (auto task : activeTasks) {
manager.cancelTask(task);
}
return 0;
default:
cout << "无效的选择" << endl;
}
}
// 清理CURL
curl_global_cleanup();
return 0;
}
2. 配置文件 (config.json)
{
"max_threads": 8,
"default_thread_count": 4,
"download_dir": "./downloads",
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"proxy": {
"enabled": false,
"type": "http",
"host": "proxy.example.com",
"port": 8080,
"username": "",
"password": ""
},
"ftp": {
"passive_mode": true,
"timeout": 30
},
"http": {
"timeout": 30,
"max_redirects": 5
}
}
3. 跨平台编译脚本 (build.sh)
#!/bin/bash
# 检测操作系统
OS=$(uname -s)
echo "Building for $OS"
# 创建构建目录
mkdir -p build
cd build
# 配置CMake
cmake .. \
-DCMAKE_BUILD_TYPE=Release \
-DCMAKE_CXX_STANDARD=17
# 编译
make -j$(nproc)
# 运行
./download_manager
功能特点
1. 多线程下载
- 将文件分成多个部分并行下载
- 可配置线程数量（默认4线程）
- 自动平衡各线程下载负载
2. 断点续传
- 记录已下载部分
- 支持暂停/继续下载
- 意外中断后可继续下载
3. 协议支持
- HTTP/HTTPS：支持Range头实现部分下载
- FTP：支持REST命令实现断点续传
- 自动处理重定向
4. 下载管理
- 任务队列管理
- 下载速度限制
- 下载进度显示
- 错误重试机制
5. 用户界面
- 命令行交互界面
- 任务状态显示
- 支持多任务管理
技术实现细节
1. 文件分块策略
void calculateParts() {
partSizes.resize(threadCount);
partDownloaded.resize(threadCount, 0);
long partSize = fileSize / threadCount;
for (int i = 0; i < threadCount; i++) {
partSizes[i] = partSize;
}
partSizes[threadCount - 1] += fileSize % threadCount;
}
2. 断点续传实现
// 检查已下载部分
long localDownloaded = 0;
int fd = open((task->savePath + ".part").c_str(), O_RDWR);
if (fd != -1) {
lseek(fd, start, SEEK_SET);
char buffer[1024];
ssize_t bytesRead;
while ((bytesRead = read(fd, buffer, sizeof(buffer))) > 0) {
localDownloaded += bytesRead;
}
close(fd);
}
// 设置Range头
if (localDownloaded > 0) {
start += localDownloaded;
task->partDownloaded[partIndex] = localDownloaded;
task->downloadedSize += localDownloaded;
}
string rangeHeader = "Range: bytes=" + to_string(start) + "-" + to_string(end);
3. 多线程同步
// 暂停/继续机制
unique_lock<mutex> lock(task->mtx);
while (task->isPaused) {
task->cv.wait(lock);
if (task->isCancelled) return 1;
}
// 任务完成通知
task->activeThreads--;
task->cv.notify_all();
4. 协议处理
// HTTP下载
curl_easy_setopt(curl, CURLOPT_URL, task->url.c_str());
curl_easy_setopt(curl, CURLOPT_RANGE, (to_string(start) + "-" + to_string(end)).c_str());
// FTP下载
curl_easy_setopt(curl, CURLOPT_URL, ("ftp://" + task->url).c_str());
curl_easy_setopt(curl, CURLOPT_USERPWD, "username:password");
使用说明
1. 编译程序
# Linux/macOS
g++ -o downloader main.cpp -lcurl -lpthread -std=c++17
# Windows (使用MinGW)
g++ -o downloader.exe main.cpp -lwininet -lws2_32 -lpthread -std=c++17
2. 运行程序
./downloader
3. 操作指南
- 选择"添加下载任务"
- 输入文件URL和保存路径
- 设置线程数（默认4）
- 使用其他选项管理下载任务
4. 命令行参数
./downloader -u <URL> -o <output_path> -t <thread_count>
参考代码 FTP、HTTP 多线程断点续传下载文件 www.youwenfan.com/contentcnt/122409.html
扩展功能
1. 添加FTP支持
void downloadFtpPart(DownloadTask* task, int partIndex) {
// 解析FTP URL
string host, path, user, pass;
parseFtpUrl(task->url, host, path, user, pass);
// 建立FTP连接
CFtpConnection* pConn = ftpSession.GetFtpConnection(host, user, pass);
// 设置恢复位置
long start = calculateStartPosition(task, partIndex);
pConn->SetCurrentDirectory(path);
CInternetFile* pFile = pConn->OpenFile(task->fileName, GENERIC_READ,
FTP_TRANSFER_TYPE_BINARY,
start);
// 下载数据
char buffer[4096];
while (!task->isCancelled) {
int bytesRead = pFile->Read(buffer, sizeof(buffer));
if (bytesRead <= 0) break;
// 写入本地文件
writeToFile(task, partIndex, buffer, bytesRead);
}
// 清理
pFile->Close();
delete pFile;
pConn->Close();
delete pConn;
}
2. 添加下载速度限制
// 在下载循环中添加
auto startTime = chrono::steady_clock::now();
size_t bytesDownloaded = 0;
while (downloading) {
// 下载数据...
bytesDownloaded += bytesRead;
// 检查速度
auto now = chrono::steady_clock::now();
auto elapsed = chrono::duration_cast<chrono::milliseconds>(now - startTime).count();
if (elapsed > 1000) { // 每秒检查一次
double speed = (bytesDownloaded * 1000.0) / elapsed;
if (speed > maxSpeed) {
sleep for a while...
}
bytesDownloaded = 0;
startTime = now;
}
}
3. 添加下载计划任务
void scheduleDownload(DownloadTask* task, time_t scheduledTime) {
thread scheduler([task, scheduledTime] {
time_t now = time(nullptr);
if (scheduledTime > now) {
sleep(scheduledTime - now);
}
// 开始下载
manager.addTask(task);
});
scheduler.detach();
}
性能优化
1. 连接复用
// 使用连接池
class ConnectionPool {
map<string, vector<CURL*>> connections;
public:
CURL* getConnection(const string& url) {
if (connections[url].empty()) {
return curl_easy_init();
}
CURL* conn = connections[url].back();
connections[url].pop_back();
return conn;
}
void releaseConnection(const string& url, CURL* conn) {
connections[url].push_back(conn);
}
};
2. 零拷贝技术
// 使用sendfile系统调用
ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count) {
// 直接在内核空间传输数据
}
3. 内存映射文件
// 使用mmap提高文件写入速度
void* mapped = mmap(NULL, fileSize, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
// 各线程直接写入映射内存
munmap(mapped, fileSize);
安全考虑
1. URL验证
bool isValidUrl(const string& url) {
regex pattern(R"(^(http|https|ftp)://[a-zA-Z0-9\-\.]+\.[a-zA-Z]{2,}(/\S*)?$)");
return regex_match(url, pattern);
}
2. 文件名校验
string sanitizeFilename(const string& filename) {
string result;
for (char c : filename) {
if (isalnum(c) || c == '.' || c == '_' || c == '-') {
result += c;
} else {
result += '_';
}
}
return result;
}
3. 证书验证
// HTTPS证书验证
curl_easy_setopt(curl, CURLOPT_SSL_VERIFYPEER, 1L);
curl_easy_setopt(curl, CURLOPT_SSL_VERIFYHOST, 2L);
curl_easy_setopt(curl, CURLOPT_CAINFO, "/path/to/cert.pem");
项目总结
这个多线程断点续传下载程序实现了以下核心功能：
-
多协议支持：
- HTTP/HTTPS：使用Range头实现部分下载
- FTP：使用REST命令实现断点续传
-
多线程下载：
- 文件分块并行下载
- 可配置线程数量
- 自动负载均衡
-
断点续传：
- 记录已下载部分
- 支持暂停/继续
- 意外中断后恢复
-
下载管理：
- 任务队列管理
- 下载速度控制
- 错误重试机制
-
用户界面：
- 命令行交互界面
- 任务状态显示
- 多任务管理
