spark-qt-research/MultiplethreadDownload/downloadworker.cpp

#include "downloadworker.h"
#include <QIODevice>
#include <QEventLoop>
#include <QNetworkAccessManager>
#include <QNetworkRequest>
#include <QNetworkReply>
#include <QDebug>

DownloadWorker::DownloadWorker(QObject *parent)
{

}

void DownloadWorker::setIdentifier(int identifier)
{
    this->identifier = identifier;
}

void DownloadWorker::setParamter(const QString &url, QPair<qint64, qint64> range, QFile *file)
{
    this->url = url;
//    this->range = range;
    this->startPos = range.first;
    this->endPos = range.second;
    this->file = file;
}

void DownloadWorker::doWork()
{
        QNetworkAccessManager *mgr = new QNetworkAccessManager(this);
        QNetworkRequest request;
        request.setUrl(url);
        request.setAttribute(QNetworkRequest::FollowRedirectsAttribute, true);
        request.setRawHeader("Range", QString("bytes=%1-%2").arg(startPos)
                    .arg(endPos).toLocal8Bit());
//        QNetworkReply *reply = mgr->get(request);
//        QNetworkReply *reply = mgr->get(request);
        reply = mgr->get(request);
//        qDebug() << "开始下载数据：" << QString(" %1~%2 -> writePos Start %3")
//            .arg(range.first).arg(range.second).arg(writePos);
        qDebug() << "开始下载数据：" << QString(" %1~%2 -> writePos Start %3")
            .arg(startPos).arg(endPos).arg(receivedPos);
//        connect(reply, &QNetworkReply::readyRead, [reply, this](){
////            qDebug() << "测试是否触发，午安啦啦啦";
//            QByteArray data = reply->readAll();
//            qDebug() << "获取到了数据" << " " << data.size();
//            emit resultReady(identifier, data);
//        });
//        connect(reply, &QNetworkReply::errorOccurred, [reply](QNetworkReply::NetworkError error){
//            if (error != QNetworkReply::NoError) {
//                qDebug() << "出错了：" << reply->errorString();
//            }
//        });
        connect(reply, &QNetworkReply::finished, mgr, &QNetworkAccessManager::deleteLater);
        connect(reply, &QNetworkReply::finished, reply, &QNetworkReply::deleteLater);
        connect(reply, &QNetworkReply::readyRead, this, &DownloadWorker::dataReady);
        connect(reply, &QNetworkReply::finished, this, &DownloadWorker::slotFinish);
}

void DownloadWorker::dataReady()
{
            qDebug() << QString("测试是否执行: %1 %2").arg(startPos).arg(receivedPos);
            QByteArray data = reply->readAll();
            qDebug() << "获取到了数据" << " " << data.size();
//            emit resultReady(identifier, data);
//            auto controller = qobject_cast<DownloadController*>(parent());
//            QMetaObject::invokeMethod(controller, "handleResults", Qt::DirectConnection,
//                                      Q_ARG(int, identifier), Q_ARG(QByteArray, data));
//            QMetaObject::invokeMethod(controller, "handleResults", Qt::QueuedConnection,
        file->seek(startPos + receivedPos);
        file->write(data);
        qDebug() << QString("%1, %2, %3").arg(startPos).arg(receivedPos).arg(data.size());
        receivedPos += data.size();
        //                                      Q_ARG(int, identifier), Q_ARG(QByteArray, data));
}

void DownloadWorker::slotFinish()
{
    file->flush();
    qDebug() << "数据块下载完毕：" << QString(" %1~%2 -> writePos Start %3")
        .arg(startPos).arg(endPos).arg(receivedPos);
}


DownloadController::DownloadController(QObject *parent)
{
    this->threadNum = 4;

    // 注册信号槽传递的数据类型
    qRegisterMetaType<QPair<qint64, qint64>>("QPair<qint64, qint64>");
}

DownloadController::~DownloadController()
{

}


void DownloadController::setFilename(QString filename)
{
    this->filename = filename;
}

void DownloadController::setThreadNum(int threadNum)
{
    this->threadNum = threadNum;
}

/**
 * @brief 开始下载
 */
void DownloadController::startDownload(const QString &url, qint64 fileSize)
{
    // 下载任务等分，计算每个线程的下载数据
    qint64 segmentSize = fileSize / threadNum;
    ranges.resize(threadNum);
//    QVector<qint64> writePosList;
    writePosList.resize(threadNum);
    for (int i = 0; i < threadNum; i++) {
        ranges[i].first = i * segmentSize;
        ranges[i].second = i * segmentSize + segmentSize - 1;
        writePosList[i]  = ranges[i].first;
    }
    ranges[threadNum-1].second = fileSize; // 余数部分加入最后一个

    qDebug() << QString("查看分段下载：%1 %2").arg(ranges.size()).arg(writePosList.size());
//    return;
    // 打开文件
    file = new QFile;
    file->setFileName(filename);
    if (file->exists())
        file->remove();
    if (!file->open(QIODevice::WriteOnly)) {
        emit errorOccur(file->errorString());
        return;
    }
    file->resize(fileSize);

    // 创建下载线程
//    qint64 bytesReceived = 0;
    for(int i = 0; i < ranges.size(); i++) {
        qDebug() << QString("第%1个线程：%2-%3").arg(i).arg(ranges.at(i).first).arg(ranges.at(i).second);
        auto worker = new DownloadWorker(this);
        auto range = ranges.at(i);
        worker->setIdentifier(i);
        worker->setParamter(url, range, file);
//        connect(worker, &DownloadWorker::resultReady, this, &DownloadController::handleResults);

        worker->doWork();
    }
}

/**
 * @brief 暂停下载
 */
void DownloadController::paruseDownload()
{

}

/**
 * @brief 停止下载
 */
void DownloadController::stopDownload()
{

}

/**
 * @brief 写入响应数据
 * @param identifier 线程编号
 * @param data 请求数据
 */
void DownloadController::handleResults(int identifier, QByteArray data)
{
    qDebug() << QString("测试哈哈哈: %1 --- %2").arg(writePosList.size()).arg(identifier);
    QMutexLocker lock(&mutex);
    qint64 writePos = writePosList.at(identifier);
    file->seek(writePos);
    file->write(data);
    qDebug() << QString("%1, %2, %3").arg(writePos).arg(bytesReceived).arg(data.size());
    writePos += data.size();
    writePosList.replace(identifier, writePos);
    bytesReceived += data.size();
    qDebug() << "已经下载了数据: " << bytesReceived;
    emit receivedProcess(bytesReceived);
}