首页 > 编程语言 >c++多线程按行读取同一个每行长度不规则文件

c++多线程按行读取同一个每行长度不规则文件

时间：2024-03-02 18:55:59浏览次数：28  
标签：读取 pos c++ next 线程 按行 file 多线程 size

对于非常大的比如上百G的大文件读取,单线程读是非常非常慢的,需要考虑用多线程读,多个线程读同一个文件时不用加锁的,每个线程打开一个独立的文件句柄

多线程读同一个文件实现思路

思路1

  1. 先打开一个文件句柄,获取整个文件大小file_size
  2. 确定要采用线程读取的部分大小read_size和多线程的个数thread_num,算出平均每个线程要读取的大小为read_size/thread_num=each_size
  3. 计算出每个线程读取的位置start_pos和它下一个线程的读取位置next_pos
  4. 对于每个线程来说,读取时的情况可以有如下种情况:
  • start_pos等于0(整个文件都采用多线程读取),这种情况下直接用getline读取,直到读取某一行后读取指针位置超过next_pos
  • start_pos>0, 读取当前位置所在的字符,如果字符恰好为\n,则直接用getline读取,直到读取某一行后读取指针位置超过next_pos
  • start_pos>0, 读取当前位置所在的字符,如果字符不为\n,则先用getline读取一行,假设读取这行后新的位置为cur_pos,如果cur_pos >= next_pos则这个线程直接退出,不读取任何数据,因为这个线程的下一个线程会和它读取同一行,这一行的内容应该有下一个线程读取; 如果cur_pos < next_pos则当前读取的这一行直接丢弃(因为这一行交给了上一个线程来读取), 直接从下一行开始用getline读取,直到读取某一行后读取指针位置超过next_pos
  • 最后代码还要计算剩下的部分,因为文件大小read_size不一定能整除线程个数thread_num,剩下的部分应该全部交给主线程来读

这个思路实现起来容易出bug,需要保证每一个线程至少能读取一个完整的行

源码实现

可能有bug,但是功能基本实现

#include "spdlog/sinks/basic_file_sink.h"
#include "spdlog/spdlog.h"
#include <chrono>
#include <fstream>
#include <iostream>
#include <sstream>
#include <string>
#include <thread>
#include <vector>
using namespace std;

void init_log()
{
    try
    {
        auto new_logger = spdlog::basic_logger_mt("new_default_logger", "test.log", true);
        spdlog::set_default_logger(new_logger);
        spdlog::info("new logger log start");
    }
    catch (const spdlog::spdlog_ex &ex)
    {
        std::cout << "Log init failed: " << ex.what() << std::endl;
    }
}

void thread_read_file(int tid, const string &file_path, std::streampos start_pos, std::streampos next_pos, int each_size)
{
    ifstream file(file_path.c_str(), ios::in);
    if (!file.good())
    {
        file.close();
        spdlog::info("线程{} 打开文件{}失败", tid, file_path);
        return;
    }

    file.seekg(start_pos, ios::beg);
    //
    string text;
    if (start_pos != 0)
    {
        char cur_ch = 0;
        // spdlog::info("读取前{}", file.tellg());
        file.read(&cur_ch, 1); //会让指针向后移动一个字节
        // spdlog::info("读取后{}", file.tellg());
        if (start_pos == 115)
        {
            spdlog::info("tid={},115={}", tid, cur_ch);
        }
        if (cur_ch != '\n')
        {
            getline(file, text);
            spdlog::info("线程{},跳过{}", tid, text);
            if (file.tellg() >= next_pos)
            {
                /*
                1. 如果线程起始位置不为换行符,则要跳过本行,本行内容交给上一个线程读取,如果跳过本行后的读取位置(一定是换行符)>=下一个线程的起始位置,
                如果位置等于下一个线程起始位置,说明下个线程起始位置是换行符,下一行内容应该由下一个线程读取;如果位置>下一个线程起始位置,同样本行内容由上一个线程
                读取,下一行内容也不用本线程读取,可能是下一个线程读取
                 */
                spdlog::info("线程{} start_pos={},next_pos={},each_size={} 起始位置不是\\n,读取一行后的指针位置{}>=next_pos,不需要读取内容",
                             tid, start_pos, next_pos, each_size, file.tellg());
                file.close();
                return;
            }
        }
        else
        {
            file.seekg(-1, ios::cur);
        }
        // spdlog::info("线程{} cur_ch={}", tid, cur_ch);
    }

    std::streampos cur_pos = file.tellg();
    while (cur_pos < next_pos && getline(file, text))
    {
        /*
        1. cur_pos始终指向每一行的行尾,如果cur_pos=next_pos则说明next_pos是行尾,则接下来的一行应该由
        下一个线程读,所以这里是cur_pos < next_pos,而不是cur_pos <= next_pos
         */
        int cur_line_len = file.tellg() - cur_pos;
        spdlog::info("线程{} start_pos={},next_pos={},each_size={},本行开始pos={},本行结束pos={},本行读长={},text={}",
                     tid, start_pos, next_pos, each_size, cur_pos, file.tellg(), cur_line_len, text);
        cur_pos = file.tellg();
    }
    spdlog::info("线程{} start_pos={},next_pos={},each_size={},结束时cur_pos={},总共区间长度为{}\n", tid, start_pos, next_pos, each_size, cur_pos, cur_pos - start_pos);
    file.close();
    return;
}

void test_detach(const string &file_path)
{
    // for (int i = 0; i < 10; ++i)
    // {
    //     std::thread th(thread_read_file, i, file_path);
    //     th.detach();
    // }
}

void test_join(const string &file_path)
{
    //确定文件长度
    ifstream file(file_path.c_str(), ios::in);

    //把指针指到文件末尾求出文件大小
    int file_size = file.seekg(0, ios::end).tellg();
    file.close();

    int thread_nums = 50;                       //线程个数
    int each_size = file_size / thread_nums;    //平均每个线程读取的字节数
    std::streampos start_pos = 0, next_pos = 0; //每个线程读取位置的起始和下一个线程读取的起始位置
    vector<std::thread> vec_threads;            //线程列表
    spdlog::info("thread_nums={},each_size={},file_size={}", thread_nums, each_size, file_size);
    int t_id = 0; //线程id
    for (; t_id < thread_nums; ++t_id)
    {
        next_pos += each_size;
        std::thread th(thread_read_file, t_id, file_path, start_pos, next_pos, each_size);
        vec_threads.emplace_back(std::move(th)); // push_back() is also OK
        start_pos = next_pos;
    }
    if (file_size % thread_nums != 0)
    {
        thread_read_file(t_id, file_path, start_pos, file_size, each_size);
    }

    for (auto &it : vec_threads)
    {
        it.join();
    }
}
int main()
{
    init_log();
    string file_path = "./1.txt";
    // test_detach(file_path);
    // std::this_thread::sleep_for(std::chrono::seconds(1)); // wait for detached threads done
    test_join(file_path);
    return 0;
}

思路2

  • 整体思路和方法1一样,只是读取的时候不是按照位置来判断每个线程应该读取多少,而是统计每个线程读取的长度
  • 每次移动位置指针时,记录一下移动的位置,因为每个线程应该读取的平均长度已经提前计算,只要线程读取的数据超过了平均大小,或者读取到了文件末尾就结束

源码实现

没有bug,可以适应多个线程被分配到同一行的情况,但是每个线程读取的大小必须>0

#include <chrono>
#include <fstream>
#include <iostream>
#include <sstream>
#include <string>
#include <thread>
#include <vector>
using namespace std;

void thread_read_file(int tid, const string &file_path, int start_pos, int next_pos, int each_size)
{
    ifstream file(file_path.c_str(), ios::in);
    if (!file.good())
    {
        stringstream ss;
        ss << "Thread " << tid << " failed to open file: " << file_path << endl;
        cout << ss.str();
        return;
    }

    file.seekg(start_pos, ios::beg);
    //
    string text;
    stringstream ss;

    if (start_pos != 0)
    {
        char cur_ch;
        file.read(&cur_ch, 1);
        // ss << "Thread " << tid << ", cur_ch=" << cur_ch << endl;
        if (cur_ch != '\n')
        {
            getline(file, text);
        }
    }

    while (getline(file, text) && start_pos <= next_pos)
    {
        ss << "Thread " << tid << ", start_pos=" << start_pos << ";next_pos="
           << next_pos << ";each_size=" << each_size << ": " << text << endl;
        cout << ss.str();
        start_pos = file.tellg();
    }
    file.close();
    return;
}

void test_detach(const string &file_path)
{
    // for (int i = 0; i < 10; ++i)
    // {
    //     std::thread th(thread_read_file, i, file_path);
    //     th.detach();
    // }
}

void test_join(const string &file_path)
{
    //确定文件长度
    ifstream file(file_path.c_str(), ios::in);

    //把指针指到文件末尾求出文件大小
    int file_size = file.seekg(0, ios::end).tellg();
    file.close();

    int thread_nums = 10;
    int each_size = file_size / thread_nums;
    int start_pos = 0, next_pos = 0;

    vector<std::thread> vec_threads;
    int t_id = 0;
    for (; t_id < thread_nums; ++t_id)
    {
        next_pos += each_size;
        std::thread th(thread_read_file, t_id, file_path, start_pos, next_pos, each_size);
        vec_threads.emplace_back(std::move(th)); // push_back() is also OK
        start_pos = next_pos;
    }
    if (file_size % thread_nums != 0)
    {
        thread_read_file(t_id, file_path, start_pos, next_pos, each_size);
    }

    for (auto &it : vec_threads)
    {
        it.join();
    }
}
int main()
{
    string file_path = "./1.txt";
    // test_detach(file_path);
    // std::this_thread::sleep_for(std::chrono::seconds(1)); // wait for detached threads done
    test_join(file_path);
    return 0;
}

运行结果

file

本文由博客一文多发平台 OpenWrite 发布！

标签：读取,pos,c++,next,线程,按行,file,多线程,size
From： https://www.cnblogs.com/zhaobulw/p/18049072

相关文章

  • c++多线程编程
    c++线程库：<thread>创建线程：需要可调用的函数或者函数对象作为线程入口点例：std::threadthreadname(function_name,args...)在C++中，当使用std::thread创建线程并传递类的成员函数时，需要使用&来获取成员函数的地址，同时还需要传递对象的指针（或引用）作为第一个参数。......
  • C++ 继承
    1//基类2classAnimal{3  //eat()函数4  //sleep()函数5};678//派生类9classDog:publicAnimal{10  //bark()函数11};基类&派生类一个类可以派生自多个类，这意味着，它可以从多个基类继承数据和函数。定义一个派生类，我......
  • C++ 类的静态成员
    原文可以使用 static 关键字来把类成员定义为静态的。当我们声明类的成员为静态时，这意味着无论创建多少个类的对象，静态成员都只有一个副本。静态成员在类的所有对象中是共享的。如果不存在其他的初始化语句，在创建第一个对象时，所有的静态数据都会被初始化为零。我们不能把静态......
  • C++ 指向类的指针
    原文一个指向C++类的指针与指向结构的指针类似，访问指向类的指针的成员，需要使用成员访问运算符 ->，就像访问指向结构的指针一样。与所有的指针一样，在使用指针之前，对指针进行初始化。在C++中，指向类的指针指向一个类的对象，与普通的指针相似，指向类的指针可以用于访问对象的成员......
  • C++ 拷贝构造函数（初学有点难理解）
    拷贝构造函数是一种特殊的构造函数，它在创建对象时，是使用同一类中之前创建的对象来初始化新创建的对象。拷贝构造函数通常用于：通过使用另一个同类型的对象来初始化新创建的对象。复制对象把它作为参数传递给函数。复制对象，并从函数返回这个对象。如果在类中没有定义拷......
  • C++填坑系列——新手写代码易错点
    c++新手写代码的几个易错点学习自https://www.youtube.com/watch?v=i_wDa2AS_8w1.限制usingnamespacestd的作用范围如果你自己定义了一个和std空间内同名的函数，当你把std放到globalnamespace（也就是直接usingnamespacestd）中，就会出现函数冲突；usingnamespacestd的使......
  • C++填坑系列——万能引用+移动语义+完美转发
    模板编程中的万能引用、移动语义、完美转发万能引用：T&&，辅助模板编程，这样左值和右值的参数都可以接收；移动语义：std::move，转换为右值，也可结合移动构造函数和移动赋值使用；完美转发：std::forward，可以保留参数的左值和右值属性，因为后续使用该参数可能还需要这个属性；万能引用万能引......
  • C++填坑系列——EffectiveModernC++之类型推导
    接下来会记录我在学习《EffectiveModernC++》的一些总结和思考。鉴于C++的知识太多了，我难以全面覆盖到学习，所以这里借此来补充和发散自己的学习心得:)以下内容由学习这个网站EffectiveModernC++的中文翻译内容得来https://cntransgroup.github.io/EffectiveModernCppChine......
  • C++填坑系列——EffectiveModernC++之新手怎么进入现代C++
    新手怎么进入现代C++1.使用auto来自动推导变量类型2.使用{}来创建变量和对象3.使用nullptr来创建空指针4.使用using代替typedef进行别名定义5.使用enumclass代替enum进行枚举定义6.使用=delete来禁止调用一个函数7.使用override来修饰继承链中的重写函数8.使用co......
  • C++填坑系列——EffectiveModernC++之特殊成员函数
    Chapter移步现代c++之特殊成员函数Item17:Understandspecialmemberfunctiongeneration总结：有必要了解各个函数什么时候自动生成；自动生成的函数有可能产生预期外的行为；特殊成员函数（编译器自动生成）：默认构造函数，析构函数，拷贝构造函数，拷贝赋值运算符，移动构造函数，移动赋......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99