c++ - 在 C++ unix 中异步写入文件

Question

我有一些长循环，我需要在每次迭代时将一些数据写入文件。问题是写入文件可能会很慢，所以我想通过异步写入来减少这需要的时间。

有谁知道这样做的好方法吗？我是否应该创建一个线程，通过写出它来消耗放入其缓冲区的任何内容(在这种情况下，单个生产者，单个消费者)？

我最感兴趣的是那些只涉及标准库 (C++11) 的解决方案。

Answer 1

在进入异步写入之前，如果您正在使用 IOStreams，您可能希望尝试避免意外刷新流，例如，不使用 std::endl 但是而是使用 '\n' 代替。由于对 IOStreams 的写入是经过缓冲的，因此可以大大提高性能。

如果这还不够，下一个问题是如何写入数据。如果正在进行大量格式化，则实际格式化可能会占用大部分时间。您可能能够将格式化推到一个单独的线程中，但这与仅仅将几个字节写入另一个线程完全不同:您需要传递一个合适的数据结构来保存要格式化的数据。不过，什么是合适的取决于您实际编写的内容。

最后，如果将缓冲区写入文件确实是瓶颈，并且您想坚持使用标准 C++ 库，那么让一个编写器线程监听来自合适的流缓冲区的缓冲区填充的队列可能是合理的，并且将缓冲区写入 std::ofstream:生产者接口(interface)将是一个 std::ostream，它可能会在缓冲区已满或何时发送固定大小的缓冲区流被刷新(为此我明确地使用 std::flush)到另一个读取监听的队列。下面是仅使用标准库设施快速实现该想法的方法:

#include <condition_variable>
#include <fstream>
#include <mutex>
#include <queue>
#include <streambuf>
#include <string>
#include <thread>
#include <vector>

struct async_buf
    : std::streambuf
{
    std::ofstream                 out;
    std::mutex                    mutex;
    std::condition_variable       condition;
    std::queue<std::vector<char>> queue;
    std::vector<char>             buffer;
    bool                          done;
    std::thread                   thread;

    void worker() {
        bool local_done(false);
        std::vector<char> buf;
        while (!local_done) {
            {
                std::unique_lock<std::mutex> guard(this->mutex);
                this->condition.wait(guard,
                                     [this](){ return !this->queue.empty()
                                                   || this->done; });
                if (!this->queue.empty()) {
                    buf.swap(queue.front());
                    queue.pop();
                }
                local_done = this->queue.empty() && this->done;
            }
            if (!buf.empty()) {
                out.write(buf.data(), std::streamsize(buf.size()));
                buf.clear();
            }
        }
        out.flush();
    }

public:
    async_buf(std::string const& name)
        : out(name)
        , buffer(128)
        , done(false)
        , thread(&async_buf::worker, this) {
        this->setp(this->buffer.data(),
                   this->buffer.data() + this->buffer.size() - 1);
    }
    ~async_buf() {
        std::unique_lock<std::mutex>(this->mutex), (this->done = true);
        this->condition.notify_one();
        this->thread.join();
    }
    int overflow(int c) {
        if (c != std::char_traits<char>::eof()) {
            *this->pptr() = std::char_traits<char>::to_char_type(c);
            this->pbump(1);
        }
        return this->sync() != -1
            ? std::char_traits<char>::not_eof(c): std::char_traits<char>::eof();
    }
    int sync() {
        if (this->pbase() != this->pptr()) {
            this->buffer.resize(std::size_t(this->pptr() - this->pbase()));
            {
                std::unique_lock<std::mutex> guard(this->mutex);
                this->queue.push(std::move(this->buffer));
            }
            this->condition.notify_one();
            this->buffer = std::vector<char>(128);
            this->setp(this->buffer.data(),
                       this->buffer.data() + this->buffer.size() - 1);
        }
        return 0;
    }

};

int main()
{
    async_buf    sbuf("async.out");
    std::ostream astream(&sbuf);
    std::ifstream in("async_stream.cpp");
    for (std::string line; std::getline(in, line); ) {
        astream << line << '\n' << std::flush;
    }
}