恋恋风辰的官方博客

文件传输设计思路
Posted on 2025-04-03 13:29:59 by 恋恋风辰 | In C++ | 原文链接

设计思路

文件传输必须满足以下几个条件：

  • 限制文件大小(不超过4G)
  • 长连接传输(效率高，支持大文件)
  • 客户端和服务器都知道传输进度，以保证支持断点续传(后续实现)
  • 先实现服务器单线程处理版本，在实现多线程处理版本

如遇问题可添加我的微信

<img src="https://cdn.llfc.club/wechat.jpg" alt="img" style="zoom: 33%;" />

也可以去我得哔哩哔哩主页查看项目视频详细讲解

B站主页 https://space.bilibili.com/271469206

客户端

客户端还是采用聊天项目客户端封装的TcpClient, 只是修改了发送逻辑

  1. //发送数据槽函数
  2. void TcpClient::slot_send_msg(quint16 id, QByteArray body)
  3. {
  4.     //如果连接异常则直接返回
  5.     if(_socket->state() != QAbstractSocket::ConnectedState){
  6.         emit sig_net_error(QString("断开连接无法发送"));
  7.         return;
  8.     }
  10.     //获取body的长度
  11.     quint32 bodyLength = body.size();
  13.     //创建字节数组
  14.     QByteArray data;
  15.     //绑定字节数组
  16.     QDataStream stream(&data, QIODevice::WriteOnly);
  17.     //设置大端模式
  18.     stream.setByteOrder(QDataStream::BigEndian);
  19.     //写入ID
  20.     stream << id;
  21.     //写入长度
  22.     stream << bodyLength;
  23.     //写入包体
  24.     data.append(body);
  26.     //发送消息
  27.      _socket->write(data);
  28. }

这里着重叙述以下，发送的格式是id + bodyLength + 文件流数据

其中id 为2字节，bodyLength为4字节，之后就是传输的文件流

https://cdn.llfc.club/1732450428990.jpg

slot_send_msg是槽函数，和 sig_send_msg信号连接

  1.  //连接 发送数据信号和槽函数
  2.  connect(this, &TcpClient::sig_send_msg, this, &TcpClient::slot_send_msg);

客户端在发送数据的时候调用

  1. void TcpClient::sendMsg(quint16 id,QByteArray data)
  2. {
  3.     //发送信号，统一交给槽函数处理，这么做的好处是多线程安全
  4.     emit sig_send_msg(id, data);
  5. }

客户端在打开文件对话框后选择文件，接下来，点击发送会将文件切分成固定大小的报文发送

  1. void MainWindow::on_uploadBtn_clicked()
  2. {
  3.     ui->uploadBtn->setEnabled(false);
  4.     // 打开文件
  5.        QFile file(_file_name);
  6.        if (!file.open(QIODevice::ReadOnly)) {
  7.            qWarning() << "Could not open file:" << file.errorString();
  8.            return;
  9.        }
  11.        // 保存当前文件指针位置
  12.        qint64 originalPos = file.pos();
  13.        QCryptographicHash hash(QCryptographicHash::Md5);
  14.        if (!hash.addData(&file)) {
  15.            qWarning() << "Failed to read data from file:" << _file_name;
  16.            return ;
  17.        }
  19.        _file_md5 = hash.result().toHex(); // 返回十六进制字符串
  21.     // 读取文件内容并发送
  22.     QByteArray buffer;
  23.     int seq = 0;
  25.     QFileInfo fileInfo(_file_name); // 创建 QFileInfo 对象
  27.     QString fileName = fileInfo.fileName(); // 获取文件名
  28.     qDebug() << "文件名是：" << fileName; // 输出文件名
  29.     int total_size = fileInfo.size();
  30.     int last_seq = 0;
  31.     if(total_size % MAX_FILE_LEN){
  32.         last_seq = (total_size/MAX_FILE_LEN)+1;
  33.     }else{
  34.         last_seq = total_size/MAX_FILE_LEN;
  35.     }
  37.     // 恢复文件指针到原来的位置
  38.     file.seek(originalPos);
  40.     while (!file.atEnd()) {
  41.         //每次读取2048字节发送
  42.         buffer = file.read(MAX_FILE_LEN);
  43.         QJsonObject jsonObj;
  44.         // 将文件内容转换为 Base64 编码（可选）
  45.         QString base64Data = buffer.toBase64();
  46.         //qDebug() << "send data is " << base64Data;
  47.         ++seq;
  48.         jsonObj["md5"] = _file_md5;
  49.         jsonObj["name"] = fileName;
  50.         jsonObj["seq"] = seq;
  51.         jsonObj["trans_size"] = buffer.size() + (seq-1)*MAX_FILE_LEN;
  52.         jsonObj["total_size"] = total_size;
  53.         if(buffer.size() < MAX_FILE_LEN){
  54.             jsonObj["last"] = 1;
  55.         }else{
  56.             jsonObj["last"] = 0;
  57.         }
  59.         jsonObj["data"]= base64Data;
  60.         jsonObj["last_seq"] = last_seq;
  61.         QJsonDocument doc(jsonObj);
  62.         auto send_data = doc.toJson();
  63.         TcpClient::Inst().sendMsg(ID_UPLOAD_FILE_REQ, send_data);
  64.         //startDelay(500);
  65.      }
  67.     //关闭文件
  68.     file.close();
  70. }

发送时数据字段分别为：

  • 文件md5 : 以后用来做断点续传校验

  • name : 文件名

  • seq: 报文序列号，类似于TCP序列号，自己定义的，服务器根据这个序列号组合数据写入文件。

  • trans_size: 当前已经传输的大小

  • total_size： 传输文件的总大小。

客户端需要接受服务器返回的消息更新进度条

  1. //接受服务器发送的信息
  2. void TcpClient::slot_ready_read()
  3. {
  4.     //读取所有数据
  5.     QByteArray data = _socket->readAll();
  7.     //将数据缓存起来
  8.     _buffer.append(data);
  10.     //处理收到的数据
  11.     processData();
  12. }

处理消息更新进度条

  1. void TcpClient::processData()
  2. {
  3.     while(_buffer.size() >= TCP_HEAD_LEN){
  4.         //先取出八字节头部
  5.         auto head_byte = _buffer.left(TCP_HEAD_LEN);
  6.         QDataStream stream(head_byte);
  7.         //设置为大端模式
  8.         stream.setByteOrder(QDataStream::BigEndian);
  9.         //读取ID
  10.         quint16 msg_id;
  11.         stream >> msg_id;
  12.         //读取长度
  13.         quint32 body_length;
  14.         stream >> body_length;
  15.         if(_buffer.size() >= TCP_HEAD_LEN+body_length){
  16.             //完整的消息体已经接受
  17.             QByteArray body = _buffer.mid(TCP_HEAD_LEN,body_length);
  18.             //去掉完整的消息包
  19.             _buffer = _buffer.mid(TCP_HEAD_LEN+body_length);
  20.             // 解析服务器发过来的消息
  21.             QJsonDocument jsonDoc = QJsonDocument::fromJson(body);
  22.             if(jsonDoc.isNull()){
  23.                 qDebug() << "Failed to create JSON doc.";
  24.                 this->_socket->close();
  25.                 return;
  26.             }
  28.             if(!jsonDoc.isObject()){
  29.                 qDebug() << "JSON is not an object.";
  30.                 this->_socket->close();
  31.                 return;
  32.             }
  33.             //qDebug() << "receive data is " << body;
  34.             // 获取 JSON 对象
  35.             QJsonObject jsonObject = jsonDoc.object();
  36.             emit sig_logic_process(msg_id, jsonObject);
  37.         }else{
  38.             //消息未完全接受，所以中断
  39.             break;
  40.         }
  41.     }
  42. }

单线程逻辑服务器

我们先讲解单线程处理收包逻辑的服务器，以后再给大家将多线程的。

服务器要配合客户端，对报文头部大小做修改

  1. //头部总长度
  2. #define HEAD_TOTAL_LEN 6
  3. //头部id长度
  4. #define HEAD_ID_LEN 2
  5. //头部数据长度
  6. #define HEAD_DATA_LEN 4
  7. // 接受队列最大个数
  8. #define MAX_RECVQUE  2000000
  9. // 发送队列最大个数
  10. #define MAX_SENDQUE 2000000

其余逻辑和我们在网络编程中讲的IocontextPool模型服务器一样

服务器收到报文头后调用LogicSystem来处理

  1. void CSession::AsyncReadBody(int total_len)
  2. {
  3.     auto self = shared_from_this();
  4.     asyncReadFull(total_len, [self, this, total_len](const boost::system::error_code& ec, std::size_t bytes_transfered) {
  5.         try {
  6.             if (ec) {
  7.                 std::cout << "handle read failed, error is " << ec.what() << endl;
  8.                 Close();
  9.                 _server->ClearSession(_session_id);
  10.                 return;
  11.             }
  13.             if (bytes_transfered < total_len) {
  14.                 std::cout << "read length not match, read [" << bytes_transfered << "] , total ["
  15.                     << total_len<<"]" << endl;
  16.                 Close();
  17.                 _server->ClearSession(_session_id);
  18.                 return;
  19.             }
  21.             memcpy(_recv_msg_node->_data , _data , bytes_transfered);
  22.             _recv_msg_node->_cur_len += bytes_transfered;
  23.             _recv_msg_node->_data[_recv_msg_node->_total_len] = '\0';
  24.             cout << "receive data is " << _recv_msg_node->_data << endl;
  25.             //此处将消息投递到逻辑队列中
  26.             LogicSystem::GetInstance()->PostMsgToQue(make_shared<LogicNode>(shared_from_this(), _recv_msg_node));
  27.             //继续监听头部接受事件
  28.             AsyncReadHead(HEAD_TOTAL_LEN);
  29.         }
  30.         catch (std::exception& e) {
  31.             std::cout << "Exception code is " << e.what() << endl;
  32.         }
  33.         });
  34. }

我们知道LogicSystem会将消息投递到队列里，然后单线程处理, 服务器LogicSystem注册上传逻辑

  1. void LogicSystem::RegisterCallBacks() {
  2.     _fun_callbacks[ID_TEST_MSG_REQ] = [this](shared_ptr<CSession> session, const short& msg_id,
  3.         const string& msg_data) {
  4.             Json::Reader reader;
  5.             Json::Value root;
  6.             reader.parse(msg_data, root);
  7.             auto data = root["data"].asString();
  8.             std::cout << "recv test data is  " << data << std::endl;
  10.             Json::Value  rtvalue;
  11.             Defer defer([this, &rtvalue, session]() {
  12.                 std::string return_str = rtvalue.toStyledString();
  13.                 session->Send(return_str, ID_TEST_MSG_RSP);
  14.                 });
  16.             rtvalue["error"] = ErrorCodes::Success;
  17.             rtvalue["data"] = data;
  18.     };
  20.     _fun_callbacks[ID_UPLOAD_FILE_REQ] = [this](shared_ptr<CSession> session, const short& msg_id,
  21.         const string& msg_data) {
  22.             Json::Reader reader;
  23.             Json::Value root;
  24.             reader.parse(msg_data, root);
  25.             auto data = root["data"].asString();
  26.             //std::cout << "recv file data is  " << data << std::endl;
  28.             Json::Value  rtvalue;
  29.             Defer defer([this, &rtvalue, session]() {
  30.                 std::string return_str = rtvalue.toStyledString();
  31.                 session->Send(return_str, ID_UPLOAD_FILE_RSP);
  32.                 });
  34.             // 解码
  35.             std::string decoded = base64_decode(data);
  37.             auto seq = root["seq"].asInt();
  38.             auto name = root["name"].asString();
  39.             auto total_size = root["total_size"].asInt();
  40.             auto trans_size = root["trans_size"].asInt();
  41.             auto file_path = ConfigMgr::Inst().GetFileOutPath();
  42.             auto file_path_str = (file_path / name).string();
  43.             std::cout << "file_path_str is " << file_path_str << std::endl;
  44.             std::ofstream outfile;
  45.             //第一个包
  46.             if (seq == 1) {
  47.                 // 打开文件，如果存在则清空，不存在则创建
  48.                 outfile.open(file_path_str, std::ios::binary | std::ios::trunc);
  49.             }
  50.             else {
  51.                 // 保存为文件
  52.                 outfile.open(file_path_str, std::ios::binary | std::ios::app);
  53.             }
  56.             if (!outfile) {
  57.                 std::cerr << "无法打开文件进行写入。" << std::endl;
  58.                 return 1;
  59.             }
  61.             outfile.write(decoded.data(), decoded.size());
  62.             if (!outfile) {
  63.                 std::cerr << "写入文件失败。" << std::endl;
  64.                 return 1;
  65.             }
  67.             outfile.close();
  68.             std::cout << "文件已成功保存为: " << name <<  std::endl;
  70.             rtvalue["error"] = ErrorCodes::Success;
  71.             rtvalue["total_size"] = total_size;
  72.             rtvalue["seq"] = seq;
  73.             rtvalue["name"] = name;
  74.             rtvalue["trans_size"] = trans_size;
  75.     };
  76. }

收到上传消息后写入文件。

多线程逻辑服务器

多线程逻辑服务器主要是为了缓解单线程接受数据造成的瓶颈，因为单线程接收数据，就会影响其他线程接收数据，所以考虑引入线程池处理收到的数据。

在多线程编程中我们讲过划分多线程设计的几种思路：

  1. 按照任务划分，将不同的任务投递给不同的线程
  2. 按照线程数轮询处理
  3. 按照递归的方式划分

很明显我们不是做二分查找之类的算法处理，所以不会采用第三种。

现在考虑第二种，如果客户端发送一个很大的文件，客户端将文件切分为几个小份发送，服务器通过iocontext池接受数据, 将接受的数据投递到线程池。

我们知道线程池处理任务是不分先后顺序的，只要投递到队列中的都会被无序取出处理。

https://cdn.llfc.club/1732945106584.jpg

会造成数据包处理的乱序，当然可以最后交给一个线程去组合，统一写入文件，这么做的一个弊端就是如果文件很大，那就要等待完全重组完成才能组合为一个统一的包，如果文件很大，这个时间就会很长，当然也可以暂时缓存这些数据，每次收到后排序组合，比较麻烦。

所以这里推荐按照任务划分。

按照任务划分就是按照不同的客户端做区分，一个客户端传输的数据按照文件名字的hash值划分给不同的线程单独处理，也就是一个线程专门处理对应的hash值的任务，这样既能保证有序，又能保证其他线程可以处理其他任务，也有概率会命中hash同样的值投递给一个队列，但也扩充了并发能力。

https://cdn.llfc.club/1732948742965.jpg

因为我们之前的逻辑处理也是单线程，所以考虑在逻辑层这里做一下解耦合，因为这个服务只是用来处理数据接受，不涉及多个连接互相访问。所以可以讲logic线程扩充为多个，按照sessionid将不同的逻辑分配给不同的线程处理。

https://cdn.llfc.club/1732952125218.jpg

多线程处理逻辑

将LogicSystem中添加多个LogicWorker用来处理逻辑

  1. typedef  function<void(shared_ptr<CSession>, const short &msg_id, const string &msg_data)> FunCallBack;
  2. class LogicSystem:public Singleton<LogicSystem>
  3. {
  4.     friend class Singleton<LogicSystem>;
  5. public:
  6.     ~LogicSystem();
  7.     void PostMsgToQue(shared_ptr < LogicNode> msg, int index);
  8. private:
  9.     LogicSystem();
  10.     std::vector<std::shared_ptr<LogicWorker> > _workers;
  11. };

实现投递逻辑

  1. LogicSystem::LogicSystem(){
  2.     for (int i = 0; i < LOGIC_WORKER_COUNT; i++) {
  3.         _workers.push_back(std::make_shared<LogicWorker>());
  4.     }
  5. }
  7. LogicSystem::~LogicSystem(){
  9. }
  11. void LogicSystem::PostMsgToQue(shared_ptr < LogicNode> msg, int index) {
  12.     _workers[index]->PostTask(msg);
  13. }

每一个LogicWorker都包含一个线程，这样LogicWorker可以在独立的线程里处理任务

  1. class LogicWorker
  2. {
  3. public:
  4.     LogicWorker();
  5.     ~LogicWorker();
  6.     void PostTask(std::shared_ptr<LogicNode> task);
  7.     void RegisterCallBacks();
  8. private:
  9.     void task_callback(std::shared_ptr<LogicNode>);
  10.     std::thread _work_thread;
  11.     std::queue<std::shared_ptr<LogicNode>> _task_que;
  12.     std::atomic<bool> _b_stop;
  13.     std::mutex  _mtx;
  14.     std::condition_variable _cv;
  15.     std::unordered_map<short, FunCallBack> _fun_callbacks;
  16. };

LogicWorker启动一个线程处理任务

  1. LogicWorker::LogicWorker():_b_stop(false)
  2. {
  3.     RegisterCallBacks();
  5.     _work_thread = std::thread([this]() {
  6.         while (!_b_stop) {
  7.             std::unique_lock<std::mutex> lock(_mtx);
  8.             _cv.wait(lock, [this]() {
  9.                 if(_b_stop) {
  10.                     return true;
  11.                 }
  13.                 if (_task_que.empty()) {
  14.                     return false;
  15.                 }
  17.                 return true;
  19.             });
  21.             if (_b_stop) {
  22.                 return;
  23.             }
  25.             auto task = _task_que.front();
  26.             task_callback(task);
  27.             _task_que.pop();
  28.         }
  29.     });
  31. }

当然要提前注册好任务

  1. void LogicWorker::RegisterCallBacks()
  2. {
  3.     _fun_callbacks[ID_TEST_MSG_REQ] = [this](shared_ptr<CSession> session, const short& msg_id,
  4.         const string& msg_data) {
  5.             Json::Reader reader;
  6.             Json::Value root;
  7.             reader.parse(msg_data, root);
  8.             auto data = root["data"].asString();
  9.             std::cout << "recv test data is  " << data << std::endl;
  11.             Json::Value  rtvalue;
  12.             Defer defer([this, &rtvalue, session]() {
  13.                 std::string return_str = rtvalue.toStyledString();
  14.                 session->Send(return_str, ID_TEST_MSG_RSP);
  15.                 });
  17.             rtvalue["error"] = ErrorCodes::Success;
  18.             rtvalue["data"] = data;
  19.     };
  21.     _fun_callbacks[ID_UPLOAD_FILE_REQ] = [this](shared_ptr<CSession> session, const short& msg_id,
  22.         const string& msg_data) {
  23.             Json::Reader reader;
  24.             Json::Value root;
  25.             reader.parse(msg_data, root);
  26.             auto seq = root["seq"].asInt();
  27.             auto name = root["name"].asString();
  28.             auto total_size = root["total_size"].asInt();
  29.             auto trans_size = root["trans_size"].asInt();
  30.             auto last = root["last"].asInt();
  31.             auto file_data = root["data"].asString();
  32.             Json::Value  rtvalue;
  33.             Defer defer([this, &rtvalue, session]() {
  34.                 std::string return_str = rtvalue.toStyledString();
  35.                 session->Send(return_str, ID_UPLOAD_FILE_RSP);
  36.                 });
  38.             // 使用 std::hash 对字符串进行哈希
  39.             std::hash<std::string> hash_fn;
  40.             size_t hash_value = hash_fn(name); // 生成哈希值
  41.             int index = hash_value % FILE_WORKER_COUNT;
  42.             std::cout << "Hash value: " << hash_value << std::endl;
  44.             FileSystem::GetInstance()->PostMsgToQue(
  45.                 std::make_shared<FileTask>(session, name, seq, total_size,
  46.                     trans_size, last, file_data),
  47.                 index
  48.             );
  50.             rtvalue["error"] = ErrorCodes::Success;
  51.             rtvalue["total_size"] = total_size;
  52.             rtvalue["seq"] = seq;
  53.             rtvalue["name"] = name;
  54.             rtvalue["trans_size"] = trans_size;
  55.             rtvalue["last"] = last;
  56.     };
  57. }

处理逻辑

  1. void LogicWorker::task_callback(std::shared_ptr<LogicNode> task)
  2. {
  3.     cout << "recv_msg id  is " << task->_recvnode->_msg_id << endl;
  4.     auto call_back_iter = _fun_callbacks.find(task->_recvnode->_msg_id);
  5.     if (call_back_iter == _fun_callbacks.end()) {
  6.         return;
  7.     }
  8.     call_back_iter->second(task->_session, task->_recvnode->_msg_id,
  9.         std::string(task->_recvnode->_data, task->_recvnode->_cur_len));
  10. }

比如对于文件上传，ID_UPLOAD_FILE_REQ就调用对应的回调，在回调函数里我们再次将要处理的任务封装好投递到文件系统

  1. FileSystem::GetInstance()->PostMsgToQue(
  2.                 std::make_shared<FileTask>(session, name, seq, total_size,
  3.                     trans_size, last, file_data),
  4.                 index
  5.             );

文件系统和逻辑系统类似，包含一堆FileWorker

  1. class FileSystem :public Singleton<FileSystem>
  2. {
  3.     friend class Singleton<FileSystem>;
  4. public:
  5.     ~FileSystem();
  6.     void PostMsgToQue(shared_ptr <FileTask> msg, int index);
  7. private:
  8.     FileSystem();
  9.     std::vector<std::shared_ptr<FileWorker>>  _file_workers;
  10. };

实现投递逻辑

  1. FileSystem::~FileSystem()
  2. {
  4. }
  6. void FileSystem::PostMsgToQue(shared_ptr<FileTask> msg, int index)
  7. {
  8.     _file_workers[index]->PostTask(msg);
  9. }
  11. FileSystem::FileSystem()
  12. {
  13.     for (int i = 0; i < FILE_WORKER_COUNT; i++) {
  14.         _file_workers.push_back(std::make_shared<FileWorker>());
  15.     }
  16. }

定义文件任务

  1. class CSession;
  2. struct FileTask {
  3.     FileTask(std::shared_ptr<CSession> session, std::string name,
  4.         int seq, int total_size, int trans_size, int last, 
  5.         std::string file_data) :_session(session),
  6.         _seq(seq),_name(name),_total_size(total_size),
  7.         _trans_size(trans_size),_last(last),_file_data(file_data)
  8.     {}
  9.     ~FileTask(){}
  10.     std::shared_ptr<CSession> _session;
  12.     int _seq ;
  13.     std::string _name ;
  14.     int _total_size ;
  15.     int _trans_size ;
  16.     int _last ;
  17.     std::string _file_data;
  18. };

实现文件工作者

  1. class FileWorker
  2. {
  3. public:
  4.     FileWorker();
  5.     ~FileWorker();
  6.     void PostTask(std::shared_ptr<FileTask> task);
  7. private:
  8.     void task_callback(std::shared_ptr<FileTask>);
  9.     std::thread _work_thread;
  10.     std::queue<std::shared_ptr<FileTask>> _task_que;
  11.     std::atomic<bool> _b_stop;
  12.     std::mutex  _mtx;
  13.     std::condition_variable _cv;
  14. };

构造函数启动线程

  1. FileWorker::FileWorker():_b_stop(false)
  2. {
  3.     _work_thread = std::thread([this]() {
  4.         while (!_b_stop) {
  5.             std::unique_lock<std::mutex> lock(_mtx);
  6.             _cv.wait(lock, [this]() {
  7.                 if (_b_stop) {
  8.                     return true;
  9.                 }
  11.                 if (_task_que.empty()) {
  12.                     return false;
  13.                 }
  15.                 return true;
  16.             });
  18.             if (_b_stop) {
  19.                 break;
  20.             }
  22.             auto task = _task_que.front();
  23.             _task_que.pop();
  24.             task_callback(task);
  25.         }
  27.     });
  28. }

析构需等待线程

  1. FileWorker::~FileWorker()
  2. {
  3.     _b_stop = true;
  4.     _cv.notify_one();
  5.     _work_thread.join();
  6. }

投递任务

  1. void FileWorker::PostTask(std::shared_ptr<FileTask> task)
  2. {
  3.     {
  4.         std::lock_guard<std::mutex> lock(_mtx);
  5.         _task_que.push(task);
  6.     }
  8.     _cv.notify_one();
  9. }

因为线程会触发回调函数保存文件，所以我们实现回调函数

  1. void FileWorker::task_callback(std::shared_ptr<FileTask> task)
  2. {
  4.     // 解码
  5.     std::string decoded = base64_decode(task->_file_data);
  7.     auto file_path = ConfigMgr::Inst().GetFileOutPath();
  8.     auto file_path_str = (file_path / task->_name).string();
  9.     auto last = task->_last;
  10.     //std::cout << "file_path_str is " << file_path_str << std::endl;
  11.     std::ofstream outfile;
  12.     //第一个包
  13.     if (task->_seq == 1) {
  14.         // 打开文件，如果存在则清空，不存在则创建
  15.         outfile.open(file_path_str, std::ios::binary | std::ios::trunc);
  16.     }
  17.     else {
  18.         // 保存为文件
  19.         outfile.open(file_path_str, std::ios::binary | std::ios::app);
  20.     }
  23.     if (!outfile) {
  24.         std::cerr << "无法打开文件进行写入。" << std::endl;
  25.         return ;
  26.     }
  28.     outfile.write(decoded.data(), decoded.size());
  29.     if (!outfile) {
  30.         std::cerr << "写入文件失败。" << std::endl;
  31.         return ;
  32.     }
  34.     outfile.close();
  35.     if (last) {
  36.         std::cout << "文件已成功保存为: " << task->_name << std::endl;
  37.     }
  39. }

测试效果

https://cdn.llfc.club/1732955339237.jpg

源码链接

https://gitee.com/secondtonone1/boostasio-learn/tree/master/network/day26-multithread-res-server

热门评论
点击查看更多评论 没有更多评论了,点击收起 展开评论

热门文章

  1. vscode搭建windows C++开发环境
    喜欢(596) 浏览(94627)

  2. 使用hexo搭建个人博客
    喜欢(533) 浏览(13457)

  3. Linux环境搭建和编码
    喜欢(594) 浏览(15134)

  4. MarkDown在线编辑器
    喜欢(514) 浏览(15250)

  5. 聊天项目(28) 分布式服务通知好友申请
    喜欢(507) 浏览(6820)

最新评论

  1. 创建项目和编译 secondtonone1:谢谢支持
  2. 可变参数模板+异步队列实现异步打印功能 熊二:学到了%%%
  3. 网络编程学习方法和图书推荐 Corleone:啥程度可以找工作
  4. 再谈单例模式 secondtonone1:是的，C++11以后返回局部static变量对象能保证线程安全了。
  5. 堆排序 secondtonone1:堆排序非常实用，定时器就是这个原理制作的。
  6. 类和对象 陈宇航:支持！！！！
  7. boost::asio之socket的创建和连接 项空月:发现一些错别字 :每隔vector存储  是不是是每个. asio::mutable_buffers_1 o或者    是不是多打了个o
  8. 聊天项目(15) 客户端实现TCP管理者 lkx:已经在&QTcpSocket::readyRead 回调函数中做了处理了的。
  9. C++ 并发三剑客future, promise和async Yunfei:大佬您好，如果这个线程池中加入的异步任务的形参如果有右值引用，这个commit中的返回类型推导和bind绑定就会出现问题，请问实际工程中，是不是不会用到这种任务，如果用到了，应该怎么解决？
  10. 无锁并发队列 TenThousandOne:_head  和 _tail  替换为原子变量。那里pop的逻辑，val = _data[h] 可以移到循环外面吗
  11. C++ 线程安全的单例模式演变 183******95:单例模式的析构函数何时运行呢? 实际测试里：无论单例模式的析构函数为私有或公有，使用智能指针和辅助回收类，两种方法都无法在main()结束前调用单例的析构函数。
  12. 面试题汇总(一) secondtonone1:看到网络上经常提问的go的问题，做了一下汇总，结合自己的经验给出的答案，如有纰漏，望指正批评。
  13. interface应用 secondtonone1:interface是万能类型，但是使用时要转换为实际类型来使用。interface丰富了go的多态特性，也降低了传统面向对象语言的耦合性。
  14. protobuf配置和使用 熊二:你可以把dll放到系统目录，也可以配置环境变量，还能把dll丢到lib里
  15. slice介绍和使用 恋恋风辰:切片作为引用类型极大的提高了数据传递的效率和性能，但也要注意切片的浅拷贝隐患，算是一把双刃剑，这世间的常态就是在两极之间寻求一种稳定。
  16. string类 WangQi888888:确实错了，应该是！isspace(sind[index]). 否则不进入循环，还是原来的字符串“some string”
  17. Qt 对话框 Spade2077:QDialog w(); //这里是不是不需要带括号
  18. 利用C11模拟伪闭包实现连接的安全回收 搁浅:看chatgpt说 直接传递 shared_from_this() 更安全 提问： socket_.async_read_some(boost::asio::buffer(data_, BUFFSIZE), // 接收客户端发生来的数据 std::bind(&Session::handle_read, this, std::placeholders::_1, std::placeholders::_2, shared_from_this())); socket_.async_read_some(boost::asio::buffer(data_, BUFFSIZE), std::bind(&Session::handle_read, shared_from_this(), std::placeholders::_1, std::placeholders::_2)); 这两种方式有区别吗？ 回答 ： 第一种方式：this 是裸指针，可能会导致生命周期问题，虽然 shared_from_this() 提供了一定的保护，但 this 依然存在风险。 第二种方式：完全使用 shared_ptr 来管理生命周期，更加安全。 通常，第二种方式更推荐使用，因为它可以确保在异步操作完成之前，Session 对象的生命周期得到完全管理，避免使用裸指针的潜在风险。
  19. 聊天项目(7) visualstudio配置grpc diablorrr:cmake文件得改一下 find_package(Boost REQUIRED COMPONENTS system filesystem)，要加上filesystem。在target_link_libraries中也同样加上
  20. 利用栅栏实现同步 Dzher:作者你好！我觉得 std::thread a(write_x); std::thread b(write_y); std::thread c(read_x_then_y); std::thread d(read_y_then_x); 这个例子中的assert fail并不会发生，原子变量设定了非relaxed内存序后一个线程的原子变量被写入，那么之后的读取一定会被同步的，c和d线程中只可能同时发生一个z++未执行的情况，最终z不是1就是2了，我测试了很多次都没有assert，请问我这个观点有什么错误，谢谢！
  21. 聊天项目(13) 重置密码功能 Doraemon:万一一个用户多个邮箱呢 有可能的
  22. 处理网络粘包问题 zyouth: //消息的长度小于头部规定的长度，说明数据未收全，则先将部分消息放到接收节点里 if (bytes_transferred < data_len) { memcpy(_recv_msg_node->_data + _recv_msg_node->_cur_len, _data + copy_len, bytes_transferred); _recv_msg_node->_cur_len += bytes_transferred; ::memset(_data, 0, MAX_LENGTH); _socket.async_read_some(boost::asio::buffer(_data, MAX_LENGTH), std::bind(&CSession::HandleRead, this, std::placeholders::_1, std::placeholders::_2, shared_self)); //头部处理完成 _b_head_parse = true; return; } 把_b_head_parse = true;放在_socket.async_read_some前面是不是更好
  23. 解决博客回复区被脚本注入的问题 secondtonone1:走到现在我忽然明白一个道理，无论工作也好生活也罢，最重要的是开心，即使一份安稳的工作不能给我带来事业上的积累也要合理的舍弃，所以我还是想去做喜欢的方向。
  24. 构造函数 secondtonone1:构造函数是类的基础知识，要着重掌握
  25. Qt MVC结构之QItemDelegate介绍 胡歌-此生不换:gpt, google
  26. 答疑汇总(thread,async源码分析) Yagus:如果引用计数为0，则会执行 future 的析构进而等待任务执行完成，那么看到的输出将是 这边应该不对吧，std::future析构只在这三种情况都满足的时候才回block: 1.共享状态是std::async 创造的(类型是_Task_async_state) 2.共享状态没有ready 3.这个future是共享状态的最后一个引用 这边共享状态类型是“_Package_state”，引用计数即使为0也不应该block啊
  27. visual studio配置boost库 一giao里我离giaogiao:请问是修改成这样吗：.\b2.exe toolset=MinGW
  28. 聊天项目(9) redis服务搭建 pro_lin:redis线程池的析构函数，除了pop出队列，还要free掉redis连接把
  29. 基于锁实现线程安全队列和栈容器 secondtonone1:我是博主，你认真学习的样子的很可爱，哈哈，我画的是链表由空变成1个的情况。其余情况和你思考的类似，只不过我用了一个无效节点表示tail的指向，最初head和tail指向的都是这个节点。

其他链接

  1. GitHub
  2. 个人博客
  3. 博客园
  4. CSDN

个人公众号

个人微信