寻求支持边下载边解压tar.gz的C/C++库及libarchive问题排查

我想用libcurl下载tar.gz文件，实现边下载边解压——每下载一段数据块就立即解压，不用等整个文件下载完再统一解压。有没有能满足这个需求的C/C++库？

我试过用libarchive解压，但处理第一个数据块时返回了truncated gzip input错误，看起来libarchive好像需要完整文件才能解压。下面是我的代码，刚接触libarchive，不确定用法对不对：

#include <iostream>
#include <vector>
#include <string>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <atomic>
#include <thread>

// libarchive
#include <archive.h>
#include <archive_entry.h>
#include <curl/curl.h>

struct mydata {
  void *buffer;
  ssize_t *size;
};
struct curldata {
    void *buffer;
    ssize_t *size;
    CURL *curl;
};

std::atomic<bool> rd(true);
struct archive *archive, *archivefd;
std::atomic<bool> start_read(false);
la_ssize_t libarchiveRead(struct archive* a, void* client_data, const void** block)
{
    if(!rd) {
    mydata *my_data = (mydata*)client_data;
    std::cout << "calling custom read(), size " << *(my_data->size) << std::endl;
    *block = my_data->buffer;
    rd=true;
    return *(my_data->size);
    }
    return 0;
}

int libarchiveClose(struct archive* a, void* client_data)
{
    std::cout << "calling custom close() for archive" << std::endl;
    mydata *my_data = (mydata*)client_data;
    delete my_data;
    return (ARCHIVE_OK);
}

int libarchiveClosefd(struct archive* a, void* client_data)
{
    std::cout << "calling custom close() for archivefd" << std::endl;
    mydata *my_data = (mydata*)client_data;
    delete my_data;
    return (ARCHIVE_OK);
}
static size_t curlWriteFunction(void *ptr, size_t size, size_t nmemb, void *write_data) {
    //size is always 1
    curldata *my_data = (curldata*)(write_data);
    *(my_data->size) = nmemb * size;
    std::cout << "calling curlWriteFunction(), size: " << size << " , nmemb: " << nmemb
    << " , my_data->size: " << *(my_data->size) << std::endl;
    memcpy(my_data->buffer, ptr, *(my_data->size));
    curl_easy_pause(my_data->curl, CURL_WRITEFUNC_PAUSE);
    rd=false;
    return (*(my_data->size));
}


 static size_t progress(void *clientp, double dltotal, double dlnow, double ultotal, double ulnow) {
    CURL *curl = (CURL *)clientp;
    (void)ultotal;
    (void)ulnow;
    if(dltotal == 0) {
        return 0;
    }
    if(rd) {

        curl_easy_pause(curl, CURLPAUSE_CONT);
        std::cout << "progress: " << dlnow/dltotal * 100 << "%" << std::endl;
    }
    return 0;
 }

void readarchive(void *client_data) {
    struct archive_entry    *entry;
    int flags = ARCHIVE_EXTRACT_TIME;
    flags |= ARCHIVE_EXTRACT_PERM;
    flags |= ARCHIVE_EXTRACT_ACL;
    flags |= ARCHIVE_EXTRACT_FFLAGS;
    while(rd);
    std::cout << "calling archive_read_open for archive.." << std::endl;
    int res = archive_read_open(archive,
                                client_data,
                                nullptr,
                                (archive_read_callback*)libarchiveRead,
                                (archive_close_callback*)libarchiveClose);
    std::cout << "called archive_read_open for archive.." << std::endl;
    res = archive_read_next_header(archive, &(entry));
    while(res == ARCHIVE_OK ) {
        std::cout << "Extracting for archive " << archive_entry_pathname(entry) << "..." << std::endl;
        // extract current entry
        archive_read_extract(archive, entry, flags);
        // read next if available
        res = archive_read_next_header(archive, &(entry));
    }
    std::cout << "archive_read_next_header for archive failed, errcode: " << res << " error: " << archive_error_string(archive) << std::endl;
}

//size_t curlWriteFunction(void *ptr, size_t size, size_t nmemb,FILE* fptr) {
//    //size is always 1
//    std::cout << "calling curlWriteFunction().." << std::endl;
//    return fwrite(ptr, size, nmemb, fptr);
//}
int main(int argc, char** argv) {

    if(argc < 3)
    {
        std::cout << argv[0] << "{-r | -w} file[s]" << std::endl;
        return 1;
    }

    std::vector<std::string> filenames;
    filenames.reserve(argc);

    while (*++argv != nullptr)
    {
        filenames.emplace_back(*argv);
    }

    bool modeRead = (filenames[0] == "-r");
    std::cout << filenames[0] << " " << filenames[1] << std::endl;

    // archive related variables

    char buff_archive[16 * 1024], buff_archivefd[16 * 1024];

    if(modeRead)
    {

        archive = archive_read_new();
        archive_read_support_filter_gzip(archive);
        archive_read_support_format_tar(archive);

        mydata *client_data = new mydata();
        int res;
        char *buff1 = new char[16 * 1024];

        client_data->size = new ssize_t;
        *(client_data->size) = 0;
        client_data->buffer = buff1;

        curldata *curl_data = new curldata();
        curl_data->size=client_data->size;
        curl_data->buffer=buff1;

        CURL *curl = curl_easy_init();
        curl_data->curl = curl;
        curl_easy_setopt(curl, CURLOPT_URL, filenames[1].c_str());
        curl_easy_setopt(curl, CURLOPT_WRITEDATA, curl_data);
        curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curlWriteFunction);
        curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0L);
        curl_easy_setopt(curl, CURLOPT_PROGRESSDATA, curl);
        curl_easy_setopt(curl, CURLOPT_PROGRESSFUNCTION,progress);
        std::thread t(readarchive, client_data);
        CURLcode result = curl_easy_perform(curl);
        if(result != CURLE_OK) {
            std::cout << "curl perform failed, errcode; " << result << " err: " << curl_easy_strerror(result) << std::endl;
        }
        t.join();
        delete client_data->size;
        delete []buff1;
        archive_read_close(archive);
        archive_read_free(archive);
        archive_read_free(archive);
        curl_easy_cleanup(curl);
    }


    return 0;

}

问题分析与解决方案 #

libarchive本身完全支持流式解压，你遇到的错误是因为代码里的数据流衔接逻辑错误：

1. 自定义libarchiveRead回调只返回了第一块数据就返回0，相当于告诉libarchive输入已经结束，导致它判定gzip数据被截断
2. 线程同步逻辑不合理，下载和解压的数据流没有持续对接，无法给libarchive提供后续的下载数据

正确的实现方式是用线程安全的缓冲区队列来衔接下载和解压线程，核心思路：

• 下载线程（curl）把收到的数据块存入队列
• 解压线程（libarchive）从队列里持续取数据，没有数据时阻塞等待
• 下载完成后标记队列结束，让解压线程处理完剩余数据再退出

以下是修改后的完整代码：

#include <iostream>
#include <vector>
#include <string>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <atomic>
#include <thread>
#include <queue>
#include <mutex>
#include <condition_variable>

#include <archive.h>
#include <archive_entry.h>
#include <curl/curl.h>

// 线程安全的下载缓冲区，用于衔接curl和libarchive
struct DownloadBuffer {
    std::queue<std::vector<char>> chunks;
    std::mutex mtx;
    std::condition_variable cv;
    bool download_finished = false;

    // 添加下载到的数据块
    void add_chunk(const char* data, size_t size) {
        std::lock_guard<std::mutex> lock(mtx);
        chunks.emplace(data, data + size);
        cv.notify_one();
    }

    // 获取数据块（无数据时阻塞，直到有新数据或下载结束）
    bool get_chunk(std::vector<char>& out) {
        std::unique_lock<std::mutex> lock(mtx);
        cv.wait(lock, [this](){ return !chunks.empty() || download_finished; });
        
        if (chunks.empty() && download_finished) {
            return false;
        }

        out.swap(chunks.front());
        chunks.pop();
        return true;
    }

    // 标记下载完成
    void finish() {
        std::lock_guard<std::mutex> lock(mtx);
        download_finished = true;
        cv.notify_one();
    }
};

// libarchive自定义读取回调：从DownloadBuffer获取数据
la_ssize_t libarchive_read_callback(struct archive* a, void* client_data, const void** block) {
    DownloadBuffer* buffer = static_cast<DownloadBuffer*>(client_data);
    static std::vector<char> current_chunk;

    if (current_chunk.empty()) {
        if (!buffer->get_chunk(current_chunk)) {
            return ARCHIVE_EOF;
        }
    }

    *block = current_chunk.data();
    la_ssize_t ret = current_chunk.size();
    current_chunk.clear();
    return ret;
}

// libarchive自定义关闭回调
int libarchive_close_callback(struct archive* a, void* client_data) {
    DownloadBuffer* buffer = static_cast<DownloadBuffer*>(client_data);
    delete buffer;
    return ARCHIVE_OK;
}

// curl自定义写入回调：把数据存入DownloadBuffer
static size_t curl_write_callback(void *ptr, size_t size, size_t nmemb, void *write_data) {
    DownloadBuffer* buffer = static_cast<DownloadBuffer*>(write_data);
    buffer->add_chunk(static_cast<const char*>(ptr), size * nmemb);
    return size * nmemb;
}

// 解压线程函数
void extract_archive_thread(DownloadBuffer* buffer) {
    struct archive* a = archive_read_new();
    archive_read_support_filter_gzip(a);
    archive_read_support_format_tar(a);

    // 设置解压选项：保留时间、权限、ACL等
    int flags = ARCHIVE_EXTRACT_TIME | ARCHIVE_EXTRACT_PERM | ARCHIVE_EXTRACT_ACL | ARCHIVE_EXTRACT_FFLAGS;

    // 绑定自定义数据源
    int res = archive_read_open(a, buffer, nullptr, libarchive_read_callback, libarchive_close_callback);
    if (res != ARCHIVE_OK) {
        std::cerr << "初始化libarchive失败: " << archive_error_string(a) << std::endl;
        archive_read_free(a);
        return;
    }

    struct archive_entry* entry;
    // 循环读取并解压每个文件
    while ((res = archive_read_next_header(a, &entry)) == ARCHIVE_OK) {
        std::cout << "正在解压: " << archive_entry_pathname(entry) << std::endl;
        archive_read_extract(a, entry, flags);
    }

    if (res != ARCHIVE_EOF) {
        std::cerr << "解压失败: " << archive_error_string(a) << std::endl;
    }

    archive_read_close(a);
    archive_read_free(a);
}

int main(int argc, char** argv) {
    if (argc != 3 || std::string(argv[1]) != "-r") {
        std::cout << "使用方式: " << argv[0] << " -r <tar.gz文件URL>" << std::endl;
        return 1;
    }

    // 初始化curl全局环境
    curl_global_init(CURL_GLOBAL_ALL);
    CURL* curl = curl_easy_init();
    if (!curl) {
        std::cerr << "初始化curl失败" << std::endl;
        return 1;
    }

    DownloadBuffer* buffer = new DownloadBuffer();

    // 设置curl下载参数
    curl_easy_setopt(curl, CURLOPT_URL, argv[2]);
    curl_easy_setopt(curl, CURLOPT_WRITEDATA, buffer);
    curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_write_callback);
    curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1L); // 跟随HTTP重定向

    // 启动解压线程
    std::thread extract_thread(extract_archive_thread, buffer);

    // 执行下载
    CURLcode result = curl_easy_perform(curl);
    if (result != CURLE_OK) {
        std::cerr << "下载失败: " << curl_easy_strerror(result) << std::endl;
    }

    // 标记下载完成，通知解压线程处理剩余数据
    buffer->finish();

    // 等待解压线程完成
    extract_thread.join();

    // 清理资源
    curl_easy_cleanup(curl);
    curl_global_cleanup();

    return 0;
}

代码关键点说明 #

• DownloadBuffer类实现了线程安全的队列，用互斥锁和条件变量保证多线程下的数据同步
• libarchive的读取回调会持续从队列取数据，直到收到ARCHIVE_EOF标记
• curl的写入回调只管将数据存入队列，无需暂停下载，提升了整体效率
• 下载完成后标记队列结束，确保解压线程能处理完所有剩余数据再退出

内容的提问来源于stack exchange，提问作者zlh121546