利用 C++ 协程实现 ‘Asynchronous Parser’：如何在处理 GB 级 JSON 的同时不阻塞主线程？

各位技术同仁，下午好！

今天，我们将深入探讨一个在现代软件开发中日益凸显的挑战：如何高效、非阻塞地处理大规模数据。具体来说，我们将以“利用 C++ 协程实现 Asynchronous Parser：在处理 GB 级 JSON 的同时不阻塞主线程”为主题，展开一场技术讲座。

在当今数据驱动的世界里，处理 GB 甚至 TB 级别的数据已是常态。无论是从网络流中解析实时数据，还是从本地文件系统加载庞大的配置或日志，解析过程的性能和响应性都至关重要。传统的同步解析方法，往往会导致主线程长时间阻塞，用户界面冻结，系统响应迟缓，这对于任何追求高性能和良好用户体验的应用来说都是不可接受的。

我们将利用 C++20 引入的协程（Coroutines）这一强大特性，构建一个异步 JSON 解析器。协程以其轻量级、非抢占式、协作式多任务的特点，为我们解决此类问题提供了优雅且高效的方案。

---

1. 问题的核心：GB 级 JSON 与主线程阻塞

想象一下，你的应用程序需要加载一个 5GB 大小的 JSON 文件。如果使用传统的同步解析库，例如 jsoncpp 或 nlohmann/json，整个文件会在一个函数调用中被读取、解析并构建成内存中的数据结构。这个过程可能需要数十秒甚至数分钟。在这段时间内：

• GUI 应用会冻结： 用户无法点击按钮，无法滚动，应用程序看起来像是崩溃了。
• 服务器应用会停止响应： 处理其他请求的线程被阻塞，导致服务质量下降。
• 实时系统会错过截止时间： 关键任务无法及时执行，可能导致严重后果。

核心痛点： I/O 操作（文件读取、网络接收）和 CPU 密集型操作（JSON 字符串解析、数据结构构建）都会耗费大量时间。在一个单线程的同步模型中，这些操作会串行执行，导致整个应用程序停滞。

---

2. 解决方案概述：C++ 协程的非阻塞魔力

为了解决主线程阻塞问题，我们通常会考虑以下几种异步编程范式：

范式	优点	缺点	适用场景
回调函数	简单直接，事件驱动	回调地狱（Callback Hell），逻辑分散，错误处理复杂	简单事件响应，如 GUI 按钮点击
Futures/Promises	任务抽象，结果可传递，链式调用	复杂任务仍需手动管理线程池，栈帧切换开销	并行计算，结果等待，如网络请求
多线程	真正并行，利用多核 CPU	同步原语（锁、条件变量）复杂，上下文切换开销大，死锁	CPU 密集型任务，需要独立处理的并发操作
C++ 协程	同步式异步代码，逻辑清晰，轻量级，非阻塞	学习曲线，需要自定义promise_type和awaitable	I/O 密集型和计算密集型任务的混合，如异步解析、网络服务

C++ 协程为我们提供了一种革命性的方式来编写异步代码：它允许我们将异步操作表达得像同步代码一样直观。通过 co_await、co_yield 和 co_return 关键字，一个函数可以在执行过程中暂停（挂起），将控制权交还给调用者，并在稍后从暂停点恢复执行，而无需阻塞调用线程。

协程如何实现非阻塞？

当一个协程执行到 co_await some_awaitable_expression; 时：

1. some_awaitable_expression 会被评估，并返回一个“Awaiter”对象。
2. Awaiter 的 await_ready() 方法被调用。如果返回 true，表示操作已完成，协程继续执行。
3. 如果 await_ready() 返回 false，协程会挂起。Awaiter 的 await_suspend() 方法被调用，它负责将当前协程的句柄（std::coroutine_handle）调度到某个执行器（如线程池或事件循环）上，以便在异步操作完成后恢复协程。
4. 协程挂起后，控制权立即返回给调用者，主线程得以继续执行其他任务。
5. 当异步操作完成时，执行器会通过之前保存的协程句柄恢复协程。
6. 协程从挂起点恢复，Awaiter 的 await_resume() 方法被调用，获取异步操作的结果。

这种机制使得我们可以在进行文件 I/O 或解析计算时，将这些耗时操作包装成 awaitable 的形式，从而在操作进行期间释放主线程。

---

3. C++20 协程基础：核心概念与机制

在深入解析器之前，我们先快速回顾 C++20 协程的核心概念。

3.1. 协程关键字

• co_return: 类似于 return，用于从协程中返回一个值或表示协程完成。
• co_yield: 类似于 yield 在生成器函数中的作用，用于从协程中产生一个值，同时挂起协程，等待下次迭代恢复。
• co_await: 用于等待一个异步操作完成。当遇到 co_await 时，协程可能挂起，直到被等待的操作完成并恢复。

3.2. 协程的幕后：promise_type 与 return_object

每个协程函数都必须有一个与之关联的“返回对象”（ReturnObject）类型，该类型内部包含一个“promise_type”的嵌套类型。promise_type 是协程与外部世界交互的接口，它定义了协程生命周期中的关键行为：

• get_return_object(): 在协程体执行前调用，用于创建并返回给调用者一个 ReturnObject。
• initial_suspend(): 协程体执行前的挂起点。返回一个 awaitable 对象，通常是 std::suspend_always（立即挂起）或 std::suspend_never（不挂起）。
• final_suspend(): 协程体执行完毕或 co_return 后调用。同样返回一个 awaitable 对象，决定协程是否在结束时挂起。
• return_value(T value) / return_void(): 当协程执行 co_return value; 或 co_return; 时调用，用于设置协程的最终结果。
• unhandled_exception(): 当协程内部抛出未捕获的异常时调用。

3.3. awaitable 概念

任何实现了特定 operator co_await 或自身就是 awaitable 类型的对象，都可以被 co_await。一个 awaitable 对象的核心是提供一个 Awaiter 对象，该对象必须实现以下方法：

• bool await_ready() const noexcept: 如果返回 true，表示操作已完成，协程不需要挂起，直接继续执行。
• void await_suspend(std::coroutine_handle<> handle) noexcept: 如果 await_ready() 返回 false，则协程挂起，此方法被调用。handle 是当前挂起协程的句柄，用于将来恢复协程。此方法负责将 handle 调度到执行器。
• decltype(auto) await_resume() noexcept: 协程恢复后调用此方法，获取异步操作的结果。

---

4. 异步 JSON 解析器的架构设计

为了实现一个异步的 GB 级 JSON 解析器，我们需要以下几个核心组件：

1. Task<T> (或 AsyncResult<T>): 作为协程的返回类型，它是一个 awaitable 对象，用于封装异步操作的最终结果。这是我们自定义的协程类型。
2. ThreadPoolExecutor: 一个后台线程池，负责执行实际的 I/O 和 CPU 密集型任务，并在任务完成后恢复相应的协程。
3. AwaitableBufferReader: 模拟异步文件/网络 I/O。它能从底层源（如文件）异步读取数据块，并在数据准备好时通知等待的协程。
4. AsyncJsonLexer: 异步词法分析器，从 AwaitableBufferReader 读取字符流，并 co_yield JSON 令牌（Token）。
5. AsyncJsonParser: 异步语法分析器，从 AsyncJsonLexer 接收令牌流，并 co_await 令牌，最终构建 JSON 结构。

我们将采取推拉结合的方式：AwaitableBufferReader 是数据的生产者，AsyncJsonLexer 是令牌的生产者，AsyncJsonParser 则是最终 JSON 值的消费者。每个环节都将是异步的。

---

5. 核心构建块实现

5.1. ThreadPoolExecutor

这是我们调度协程恢复的执行器。它是一个简单的线程池，用于将 std::function<void()> 任务提交到后台线程执行。

#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future> // For std::packaged_task

class ThreadPoolExecutor {
public:
    ThreadPoolExecutor(size_t num_threads) : stop_(false) {
        for (size_t i = 0; i < num_threads; ++i) {
            workers_.emplace_back([this] {
                while (true) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(queue_mutex_);
                        condition_.wait(lock, [this] {
                            return stop_ || !tasks_.empty();
                        });
                        if (stop_ && tasks_.empty()) {
                            return;
                        }
                        task = std::move(tasks_.front());
                        tasks_.pop();
                    }
                    task();
                }
            });
        }
    }

    ~ThreadPoolExecutor() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            stop_ = true;
        }
        condition_.notify_all();
        for (std::thread& worker : workers_) {
            worker.join();
        }
    }

    // Post a task to be executed on a thread pool thread
    void post(std::function<void()> task) {
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
            tasks_.emplace(std::move(task));
        }
        condition_.notify_one();
    }

private:
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex queue_mutex_;
    std::condition_variable condition_;
    bool stop_;
};

// Global executor for simplicity in this example
// In a real application, you'd manage its lifetime carefully
ThreadPoolExecutor g_executor(std::thread::hardware_concurrency());

5.2. Task<T>：自定义协程返回类型

Task<T> 将是我们的异步操作的句柄。它是一个 awaitable 对象，并且内部包含 promise_type 的定义。

#include <coroutine>
#include <exception>
#include <optional>
#include <type_traits> // For std::is_void_v

// Forward declaration
template <typename T>
struct Task;

// Promise type for Task<T>
template <typename T>
struct TaskPromise {
    // Member to hold the result (or exception)
    std::optional<T> value_;
    std::exception_ptr exception_;
    std::coroutine_handle<> continuation_; // Handle to resume the awaiting coroutine

    // 1. get_return_object: Called by the compiler to get the return object
    Task<T> get_return_object() noexcept;

    // 2. initial_suspend: Called before the coroutine body executes
    // We want to suspend initially so the caller can co_await the Task
    std::suspend_always initial_suspend() noexcept { return {}; }

    // 3. final_suspend: Called after the coroutine body finishes (co_return or exception)
    // We suspend finally to allow the caller to retrieve the result
    struct FinalAwaiter {
        bool await_ready() const noexcept { return false; }
        void await_suspend(std::coroutine_handle<TaskPromise<T>> handle) noexcept {
            // Resume the continuation if it exists
            if (handle.promise().continuation_) {
                handle.promise().continuation_.resume();
            }
        }
        void await_resume() noexcept {}
    };
    FinalAwaiter final_suspend() noexcept { return {}; }

    // 4. return_value / return_void: Called by co_return
    void return_value(T value) noexcept { value_.emplace(std::move(value)); }
    void return_void() noexcept { /* For Task<void> */ }

    // 5. unhandled_exception: Called if an exception escapes the coroutine body
    void unhandled_exception() noexcept { exception_ = std::current_exception(); }

    // Helper to get the result (or rethrow exception)
    T get_result() {
        if (exception_) {
            std::rethrow_exception(exception_);
        }
        return std::move(value_.value()); // Assumes value_ is set
    }
};

// Specialization for Task<void>
template <>
struct TaskPromise<void> {
    std::exception_ptr exception_;
    std::coroutine_handle<> continuation_;

    Task<void> get_return_object() noexcept;
    std::suspend_always initial_suspend() noexcept { return {}; }

    struct FinalAwaiter {
        bool await_ready() const noexcept { return false; }
        void await_suspend(std::coroutine_handle<TaskPromise<void>> handle) noexcept {
            if (handle.promise().continuation_) {
                handle.promise().continuation_.resume();
            }
        }
        void await_resume() noexcept {}
    };
    FinalAwaiter final_suspend() noexcept { return {}; }

    void return_void() noexcept {}
    void unhandled_exception() noexcept { exception_ = std::current_exception(); }

    void get_result() {
        if (exception_) {
            std::rethrow_exception(exception_);
        }
    }
};

// Task<T> definition
template <typename T>
struct Task {
    using promise_type = TaskPromise<T>;
    std::coroutine_handle<promise_type> handle_;

    Task(std::coroutine_handle<promise_type> h) : handle_(h) {}
    Task(Task&& other) noexcept : handle_(std::exchange(other.handle_, nullptr)) {}
    Task& operator=(Task&& other) noexcept {
        if (this != &other) {
            if (handle_) handle_.destroy();
            handle_ = std::exchange(other.handle_, nullptr);
        }
        return *this;
    }
    ~Task() {
        if (handle_) {
            handle_.destroy();
        }
    }

    // Task is non-copyable
    Task(const Task&) = delete;
    Task& operator=(const Task&) = delete;

    // Awaiter for co_await Task<T>
    struct Awaiter {
        std::coroutine_handle<promise_type> handle_;

        bool await_ready() const noexcept {
            return handle_.done(); // If coroutine is already done, no need to suspend
        }

        void await_suspend(std::coroutine_handle<> awaiting_handle) noexcept {
            // Store the awaiting coroutine's handle so the Task can resume it
            handle_.promise().continuation_ = awaiting_handle;
            // The Task coroutine is initially suspended, so resume it to start execution
            // We use the executor to resume it on a worker thread
            g_executor.post([this] {
                handle_.resume();
            });
        }

        T await_resume() {
            return handle_.promise().get_result();
        }
    };

    Awaiter operator co_await() const & noexcept {
        return Awaiter{handle_};
    }

    // For Task<void> specialization
    void sync_wait() {
        // Simple blocking wait for demonstration. In real async, you'd integrate with event loop.
        if (!handle_.done()) {
            // This is simplified. A real sync_wait would need to pump an event loop
            // or use a more sophisticated mechanism. For now, assume a Task is eventually resumed.
            handle_.resume(); // Force initial resume if not started
        }
        handle_.promise().get_result();
    }
};

// Definition for promise_type::get_return_object
template <typename T>
Task<T> TaskPromise<T>::get_return_object() noexcept {
    return Task<T>{std::coroutine_handle<TaskPromise<T>>::from_promise(*this)};
}

Task<void> TaskPromise<void>::get_return_object() noexcept {
    return Task<void>{std::coroutine_handle<TaskPromise<void>>::from_promise(*this)};
}

解释：

• Task<T> 是一个简单的封装，持有 std::coroutine_handle<promise_type>。
• promise_type 是协程与外部世界沟通的桥梁。它负责存储协程的结果（value_）或异常（exception_），以及记录需要被恢复的调用者协程（continuation_）。
• initial_suspend() 返回 std::suspend_always 意味着协程在创建后会立即挂起，等待 co_await 它的调用者来“启动”它。
• final_suspend() 返回 FinalAwaiter，它会在协程完成时挂起，并负责恢复 continuation_，即之前 co_await 此 Task 的协程。
• operator co_await() 使得 Task<T> 对象本身可以被 co_await。其 await_suspend 方法至关重要：它将当前协程的 handle_ 提交给 g_executor，从而在后台线程上恢复 Task 协程的执行，同时将 awaiting_handle 存储在 Task 的 promise_type 中，以便 Task 完成后能恢复等待它的协程。

5.3. AwaitableBufferReader：模拟异步 I/O

我们将模拟一个可以异步读取数据块的组件。在真实世界中，这会是对 liburing、Boost.ASIO 或 Windows Overlapped I/O 的封装。这里为了简洁，我们用一个阻塞的 std::ifstream 读文件，但将其包装在一个 Task 中，从而在后台线程执行读取，不阻塞主线程。

#include <fstream>
#include <vector>
#include <string>
#include <stdexcept>
#include <iostream>

// Represents a chunk of data, could be std::string_view for zero-copy
struct BufferChunk {
    std::vector<char> data;
    size_t size; // Actual valid data size in buffer
};

class AwaitableBufferReader {
public:
    AwaitableBufferReader(std::string filename, size_t buffer_size = 4096)
        : filename_(std::move(filename)), buffer_size_(buffer_size), file_(filename_, std::ios::binary) {
        if (!file_.is_open()) {
            throw std::runtime_error("Could not open file: " + filename_);
        }
        current_chunk_.data.resize(buffer_size_);
        current_chunk_.size = 0;
        offset_ = 0;
    }

    // Asynchronously read the next chunk of data
    Task<BufferChunk> read_chunk() {
        // If there's already data in the current_chunk that hasn't been consumed, return it
        // This simple example just fetches next. Real implementation might need more sophisticated buffer management.

        // Simulate async I/O by offloading the actual read to a thread pool
        co_return co_await g_executor.post_task<BufferChunk>([this]() {
            BufferChunk chunk;
            chunk.data.resize(buffer_size_);
            file_.read(chunk.data.data(), buffer_size_);
            chunk.size = file_.gcount();
            return chunk;
        });
    }

private:
    std::string filename_;
    size_t buffer_size_;
    std::ifstream file_;
    BufferChunk current_chunk_; // For internal buffer management, if needed
    size_t offset_; // Current read offset within the chunk
};

// Add a helper to ThreadPoolExecutor to post tasks that return values
// This is a common pattern for integrating futures/promises with coroutines
template<typename T>
Task<T> ThreadPoolExecutor::post_task(std::function<T()> func) {
    struct Awaiter {
        std::packaged_task<T()> task_;
        std::future<T> future_;
        std::coroutine_handle<> awaiting_handle_; // The coroutine to resume

        Awaiter(std::function<T()> f) : task_(std::move(f)), future_(task_.get_future()) {}

        bool await_ready() const noexcept { return future_.wait_for(std::chrono::seconds(0)) == std::future_status::ready; }

        void await_suspend(std::coroutine_handle<> handle) noexcept {
            awaiting_handle_ = handle;
            // Post the actual execution of the packaged_task to the thread pool
            g_executor.post([this] {
                task_(); // Execute the actual function
                awaiting_handle_.resume(); // Resume the awaiting coroutine when done
            });
        }

        T await_resume() {
            return future_.get(); // Get the result (will block if not ready, but await_suspend handles that)
        }
    };
    co_return co_await Awaiter(std::move(func));
}

解释：

• AwaitableBufferReader 封装了文件读取。
• read_chunk() 方法返回 Task<BufferChunk>，它内部使用 g_executor.post_task 将文件读取操作推送到线程池执行。
• ThreadPoolExecutor::post_task 是一个辅助方法，它接收一个 std::function<T()>，将其包装成 std::packaged_task，然后将 packaged_task 的执行提交到线程池。当 packaged_task 完成时，它会恢复等待它的协程。这样，文件读取操作就在后台线程上完成，主线程不会被阻塞。

5.4. AsyncJsonLexer：异步词法分析器

词法分析器负责将字符流分解为有意义的 JSON 令牌（Token）。它需要 co_await AwaitableBufferReader 来获取数据，并 co_yield 令牌。

#include <string_view>
#include <variant> // For std::variant to hold different token types
#include <vector>

// Define JSON Token types
enum class JsonTokenType {
    L_BRACE, R_BRACE,      // { }
    L_BRACKET, R_BRACKET,  // [ ]
    COLON, COMMA,          // : ,
    STRING, NUMBER,        // "value", 123.45
    TRUE, FALSE, NULL_VAL, // true, false, null
    END_OF_FILE
};

// Represents a JSON token
struct JsonToken {
    JsonTokenType type;
    std::string_view value; // For STRING and NUMBER, or keyword representation
};

class AsyncJsonLexer {
public:
    AsyncJsonLexer(AwaitableBufferReader& reader) : reader_(reader), current_buffer_idx_(0) {}

    // Asynchronously yield the next JSON token
    // This is a generator coroutine
    Task<JsonToken> next_token() {
        while (true) {
            // Ensure we have data in the buffer
            if (current_buffer_idx_ >= current_chunk_.size) {
                current_chunk_ = co_await reader_.read_chunk();
                current_buffer_idx_ = 0;
                if (current_chunk_.size == 0) {
                    co_return JsonToken{JsonTokenType::END_OF_FILE, ""};
                }
            }

            // Skip whitespace
            while (current_buffer_idx_ < current_chunk_.size && is_whitespace(current_chunk_.data[current_buffer_idx_])) {
                current_buffer_idx_++;
            }

            if (current_buffer_idx_ >= current_chunk_.size) {
                // If only whitespace was left, try to read another chunk
                continue;
            }

            char c = current_chunk_.data[current_buffer_idx_];
            current_buffer_idx_++; // Consume the char

            switch (c) {
                case '{': co_return JsonToken{JsonTokenType::L_BRACE, "{"};
                case '}': co_return JsonToken{JsonTokenType::R_BRACE, "}"};
                case '[': co_return JsonToken{JsonTokenType::L_BRACKET, "["};
                case ']': co_return JsonToken{JsonTokenType::R_BRACKET, "]"};
                case ':': co_return JsonToken{JsonTokenType::COLON, ":"};
                case ',': co_return JsonToken{JsonTokenType::COMMA, ","};
                case '"': co_return co_await parse_string();
                case 't': co_return co_await parse_keyword("true", JsonTokenType::TRUE);
                case 'f': co_return co_await parse_keyword("false", JsonTokenType::FALSE);
                case 'n': co_return co_await parse_keyword("null", JsonTokenType::NULL_VAL);
                default:
                    if (std::isdigit(c) || c == '-') {
                        co_return co_await parse_number(c);
                    }
                    throw std::runtime_error("Unexpected character in JSON: " + std::string(1, c));
            }
        }
    }

private:
    AwaitableBufferReader& reader_;
    BufferChunk current_chunk_;
    size_t current_buffer_idx_;

    bool is_whitespace(char c) {
        return c == ' ' || c == 't' || c == 'n' || c == 'r';
    }

    // Helper to read characters from the buffer, potentially fetching new chunks
    Task<char> read_char() {
        if (current_buffer_idx_ >= current_chunk_.size) {
            current_chunk_ = co_await reader_.read_chunk();
            current_buffer_idx_ = 0;
            if (current_chunk_.size == 0) {
                throw std::runtime_error("Unexpected end of file while reading character");
            }
        }
        co_return current_chunk_.data[current_buffer_idx_++];
    }

    // Helper to peek character without consuming
    Task<char> peek_char() {
        if (current_buffer_idx_ >= current_chunk_.size) {
            // Need to read ahead, but not consume. This is tricky with simple BufferChunk.
            // For simplicity, we'll re-read the chunk if needed, assuming small peeks.
            // A real lexer would have a larger internal buffer with lookahead.
            co_await reader_.read_chunk(); // This might change current_chunk_
            current_buffer_idx_ = 0;
            if (current_chunk_.size == 0) {
                throw std::runtime_error("Unexpected end of file while peeking character");
            }
        }
        co_return current_chunk_.data[current_buffer_idx_];
    }

    Task<JsonToken> parse_string() {
        std::string s;
        while (true) {
            char c = co_await read_char();
            if (c == '"') {
                break;
            }
            if (c == '\') { // Handle escape sequences
                c = co_await read_char();
                switch (c) {
                    case '"': s += '"'; break;
                    case '\': s += '\'; break;
                    case '/': s += '/'; break;
                    case 'b': s += 'b'; break;
                    case 'f': s += 'f'; break;
                    case 'n': s += 'n'; break;
                    case 'r': s += 'r'; break;
                    case 't': s += 't'; break;
                    case 'u': // Unicode escape uXXXX
                        // This would require reading 4 hex chars and converting.
                        // For simplicity, we'll just add the 'u' for now and skip 4 chars
                        s += '\'; s += 'u';
                        for(int i=0; i<4; ++i) co_await read_char();
                        break;
                    default: s += c; break; // Unknown escape sequence
                }
            } else {
                s += c;
            }
        }
        co_return JsonToken{JsonTokenType::STRING, s};
    }

    Task<JsonToken> parse_number(char first_char) {
        std::string num_str;
        num_str += first_char;
        while (true) {
            char c = co_await peek_char();
            if (std::isdigit(c) || c == '.' || c == 'e' || c == 'E' || c == '+' || c == '-') {
                num_str += co_await read_char(); // Consume
            } else {
                break;
            }
        }
        co_return JsonToken{JsonTokenType::NUMBER, num_str};
    }

    Task<JsonToken> parse_keyword(const std::string& expected, JsonTokenType type) {
        // We already consumed the first char.
        std::string keyword_str;
        keyword_str += expected[0];
        for (size_t i = 1; i < expected.length(); ++i) {
            char c = co_await read_char();
            if (c != expected[i]) {
                throw std::runtime_error("Malformed keyword: expected '" + expected + "'");
            }
            keyword_str += c;
        }
        co_return JsonToken{type, expected};
    }
};

解释：

• AsyncJsonLexer 的核心是 next_token() 方法，它返回 Task<JsonToken>。
• 当需要更多数据时，它会 co_await reader_.read_chunk()，从而挂起自身，等待文件读取完成。
• parse_string()、parse_number() 等辅助函数同样是协程，它们会 co_await read_char() 和 peek_char()，这些操作内部也会处理数据缓冲和异步读取。
• 这里 string_view 用于 JsonToken 的 value 字段，以避免不必要的拷贝，但对于 parse_string 这种需要构建字符串的场景，我们仍然使用了 std::string。对于 GB 级 JSON，零拷贝解析至关重要，这通常意味着 JsonToken::value 应该直接指向原始缓冲区中的数据。我们的 BufferChunk 内部是 std::vector<char>，所以 std::string_view 是可行的，只要我们保证 BufferChunk 的生命周期长于 JsonToken。

5.5. AsyncJsonParser：异步语法分析器

语法分析器从词法分析器获取令牌，并根据 JSON 语法规则构建 JSON 值。这里我们将构建一个简单的 JsonValue 结构来表示 JSON 数据。

#include <map>

// Representing JSON values
struct JsonValue {
    enum Type {
        Null, Bool, Number, String, Array, Object
    };
    Type type;
    std::variant<
        std::monostate,
        bool,
        double, // Or std::string for exact number representation
        std::string,
        std::vector<JsonValue>,
        std::map<std::string, JsonValue>
    > data;

    // Constructors for convenience
    JsonValue() : type(Null), data(std::monostate{}) {}
    JsonValue(bool b) : type(Bool), data(b) {}
    JsonValue(double d) : type(Number), data(d) {}
    JsonValue(std::string s) : type(String), data(std::move(s)) {}
    JsonValue(std::vector<JsonValue> arr) : type(Array), data(std::move(arr)) {}
    JsonValue(std::map<std::string, JsonValue> obj) : type(Object), data(std::move(obj)) {}

    // Debug print
    std::string to_string(int indent = 0) const {
        std::string result;
        std::string prefix(indent * 2, ' ');
        switch (type) {
            case Null: result += "null"; break;
            case Bool: result += std::get<bool>(data) ? "true" : "false"; break;
            case Number: result += std::to_string(std::get<double>(data)); break;
            case String: result += """ + std::get<std::string>(data) + """; break;
            case Array: {
                result += "[n";
                for (const auto& item : std::get<std::vector<JsonValue>>(data)) {
                    result += prefix + "  " + item.to_string(indent + 1) + ",n";
                }
                if (!std::get<std::vector<JsonValue>>(data).empty()) {
                    result.pop_back(); // Remove last comma
                    result.pop_back(); // Remove last newline
                }
                result += "n" + prefix + "]";
                break;
            }
            case Object: {
                result += "{n";
                for (const auto& pair : std::get<std::map<std::string, JsonValue>>(data)) {
                    result += prefix + "  "" + pair.first + "": " + pair.second.to_string(indent + 1) + ",n";
                }
                if (!std::get<std::map<std::string, JsonValue>>(data).empty()) {
                    result.pop_back();
                    result.pop_back();
                }
                result += "n" + prefix + "}";
                break;
            }
        }
        return result;
    }
};

class AsyncJsonParser {
public:
    AsyncJsonParser(AsyncJsonLexer& lexer) : lexer_(lexer) {}

    Task<JsonValue> parse() {
        JsonToken token = co_await lexer_.next_token();
        JsonValue value = co_await parse_value(token);
        // Ensure no extra tokens remain
        JsonToken eof_check = co_await lexer_.next_token();
        if (eof_check.type != JsonTokenType::END_OF_FILE) {
            throw std::runtime_error("Unexpected token after root JSON value.");
        }
        co_return value;
    }

private:
    AsyncJsonLexer& lexer_;

    Task<JsonValue> parse_value(JsonToken token) {
        switch (token.type) {
            case JsonTokenType::L_BRACE: return co_await parse_object();
            case JsonTokenType::L_BRACKET: return co_await parse_array();
            case JsonTokenType::STRING: return JsonValue(std::string(token.value));
            case JsonTokenType::NUMBER: return JsonValue(std::stod(std::string(token.value)));
            case JsonTokenType::TRUE: return JsonValue(true);
            case JsonTokenType::FALSE: return JsonValue(false);
            case JsonTokenType::NULL_VAL: return JsonValue();
            default:
                throw std::runtime_error("Unexpected token type: " + std::string(token.value));
        }
    }

    Task<JsonValue> parse_object() {
        std::map<std::string, JsonValue> obj;
        JsonToken token = co_await lexer_.next_token();
        if (token.type == JsonTokenType::R_BRACE) {
            co_return JsonValue(obj); // Empty object
        }

        while (true) {
            if (token.type != JsonTokenType::STRING) {
                throw std::runtime_error("Expected string key in object, got: " + std::string(token.value));
            }
            std::string key = std::string(token.value);

            token = co_await lexer_.next_token();
            if (token.type != JsonTokenType::COLON) {
                throw std::runtime_error("Expected ':' after key, got: " + std::string(token.value));
            }

            token = co_await lexer_.next_token();
            JsonValue value = co_await parse_value(token);
            obj[key] = std::move(value);

            token = co_await lexer_.next_token();
            if (token.type == JsonTokenType::R_BRACE) {
                break;
            }
            if (token.type != JsonTokenType::COMMA) {
                throw std::runtime_error("Expected ',' or '}' in object, got: " + std::string(token.value));
            }
            token = co_await lexer_.next_token(); // Advance to the next key token
        }
        co_return JsonValue(obj);
    }

    Task<JsonValue> parse_array() {
        std::vector<JsonValue> arr;
        JsonToken token = co_await lexer_.next_token();
        if (token.type == JsonTokenType::R_BRACKET) {
            co_return JsonValue(arr); // Empty array
        }

        while (true) {
            JsonValue value = co_await parse_value(token);
            arr.push_back(std::move(value));

            token = co_await lexer_.next_token();
            if (token.type == JsonTokenType::R_BRACKET) {
                break;
            }
            if (token.type != JsonTokenType::COMMA) {
                throw std::runtime_error("Expected ',' or ']' in array, got: " + std::string(token.value));
            }
            token = co_await lexer_.next_token(); // Advance to the next value token
        }
        co_return JsonValue(arr);
    }
};

解释：

• AsyncJsonParser 包含 parse()、parse_object() 和 parse_array() 等递归下降解析方法。
• 每个方法都返回 Task<JsonValue>，并且在需要下一个令牌时，会 co_await lexer_.next_token()。
• 这种递归调用和 co_await 的结合，使得解析逻辑看起来像是同步的，但实际上是完全异步的。当 lexer_.next_token() 挂起时，parse_object() 或 parse_array() 也会挂起，并将控制权交回给调用者，最终回到主线程。

---

6. 整合与演示：非阻塞解析

现在，我们来编写 main 函数，将所有组件整合起来，并演示非阻塞行为。

#include <chrono>
#include <thread>
#include <iostream>
#include <filesystem> // For creating dummy file

// Function to simulate other work on the main thread
void do_other_work(int iterations, const std::string& message) {
    for (int i = 0; i < iterations; ++i) {
        std::cout << message << " " << i << std::endl;
        std::this_thread::sleep_for(std::chrono::milliseconds(10)); // Simulate some work
    }
}

// Helper to create a dummy large JSON file
void create_large_json_file(const std::string& filename, size_t min_bytes) {
    std::ofstream ofs(filename, std::ios::binary);
    if (!ofs.is_open()) {
        throw std::runtime_error("Failed to create dummy JSON file.");
    }

    ofs << "[n";
    size_t current_bytes = 2; // for "[n"

    for (int i = 0; current_bytes < min_bytes; ++i) {
        std::string entry = "  {"id": " + std::to_string(i) + ", "name": "item_" + std::to_string(i) + "_long_name_with_some_details", "value": " + std::to_string(i * 10.5) + "},n";
        ofs << entry;
        current_bytes += entry.length();
    }

    // Replace last comma with closing bracket
    ofs.seekp(-2, std::ios_base::end);
    ofs << "n]";
    ofs.close();

    std::cout << "Created dummy JSON file: " << filename << " (approx. " << std::filesystem::file_size(filename) << " bytes)" << std::endl;
}

int main() {
    std::cout << "Starting Asynchronous JSON Parser Demo" << std::endl;

    const std::string json_filename = "large_data.json";
    const size_t target_file_size_mb = 100; // Create 100MB dummy file for test
    create_large_json_file(json_filename, target_file_size_mb * 1024 * 1024);

    try {
        AwaitableBufferReader reader(json_filename, 4096); // Read 4KB chunks
        AsyncJsonLexer lexer(reader);
        AsyncJsonParser parser(lexer);

        std::cout << "nMain thread: Starting asynchronous parsing task." << std::endl;

        // Start the parsing task
        Task<JsonValue> parse_task = parser.parse();

        // While parsing happens, the main thread can do other work
        std::cout << "Main thread: Doing other important work..." << std::endl;
        do_other_work(50, "Main thread busy...");

        std::cout << "nMain thread: Waiting for parsing to complete." << std::endl;

        // Synchronously wait for the parsing task to complete (this is where we'd get the result)
        // In a real UI app, this would be an event loop polling for completion or another async step.
        // For demonstration, `sync_wait` will block until the result is ready.
        JsonValue parsed_json = parse_task.sync_wait();

        std::cout << "nMain thread: Parsing complete! Result obtained." << std::endl;
        // std::cout << "Parsed JSON: " << parsed_json.to_string() << std::endl; // For very large JSON, this will print a lot.
        std::cout << "Root JSON type: " << parsed_json.type << std::endl;
        if (parsed_json.type == JsonValue::Type::Array) {
            std::cout << "Array size: " << std::get<std::vector<JsonValue>>(parsed_json.data).size() << " elements." << std::endl;
        }

        std::cout << "nMain thread: All done." << std::endl;

    } catch (const std::exception& e) {
        std::cerr << "Error: " << e.what() << std::endl;
    }

    // Ensure executor cleans up
    // g_executor will be destroyed when main exits, joining its threads.

    return 0;
}

编译与运行：

你需要一个支持 C++20 协程的编译器（如 GCC 10+ 或 Clang 10+）。
编译命令示例：
g++ -std=c++20 -fcoroutines -o async_parser main.cpp ThreadPoolExecutor.cpp AwaitableBufferReader.cpp AsyncJsonLexer.cpp AsyncJsonParser.cpp Task.cpp -pthread
(注意：这里为了简化，所有代码都写在一个文件里，实际编译时你需要根据文件结构调整)

如果所有代码在 main.cpp 中，则：
g++ -std=c++20 -fcoroutines -o async_parser main.cpp -pthread

运行：
./async_parser

预期输出：

你会看到 Main thread busy... 的消息与解析器内部的工作交替打印。这清晰地表明，尽管文件解析工作在后台进行，主线程并没有被阻塞，它仍然能够执行其他任务。当主线程最终 co_await parse_task 完成时，它会获取解析结果。

---

7. 进一步的考量与最佳实践

7.1. 零拷贝与内存效率

对于 GB 级 JSON，每次字符串拷贝都是巨大的开销。当前的 JsonToken::value 使用 std::string_view，这是很好的。但 parse_string 方法仍然构建了 std::string。真正的零拷贝解析器会直接在原始文件缓冲区中查找字符串的起始和结束位置，并返回 std::string_view。

JsonValue 的 std::string 成员和 std::map<std::string, JsonValue> 也会导致大量内存分配。考虑使用：

• Arena Allocator (内存池): 为所有的 JsonValue 和字符串分配预先申请的大块内存，减少碎片和系统调用开销。
• std::string_view for object keys: 如果原始数据生命周期可控，用 std::string_view 作 std::map 的键。
• 自定义容器: 避免 std::vector 和 std::map 在每次插入时可能导致的重分配。

7.2. 错误处理与取消

• 异常传播： 协程内部抛出的异常会被 promise_type::unhandled_exception() 捕获，并存储在 exception_ptr 中。当 await_resume() 被调用时，如果 exception_ptr 非空，异常会被重新抛出。这使得错误处理与同步代码一致。
• 取消机制： 协程的取消是一个复杂的话题。通常需要引入一个共享的 std::stop_source/std::stop_token 或自定义的取消标志。在每个 co_await 点之前检查取消状态，如果已取消，则立即 co_return 或抛出 std::operation_canceled 异常。

7.3. 性能优化

• 缓冲区大小： 调整 AwaitableBufferReader 的 buffer_size。太小会增加 I/O 请求次数，太大可能浪费内存。通常几 KB 到几十 KB 是一个好的起点。
• 解析算法： 递归下降解析器虽然直观，但对于某些结构可能导致深层递归。对于超大 JSON，考虑基于栈的迭代解析器。
• JIT/SIMD 加速： 对于字符串解析、数字转换等核心操作，可以考虑使用 SIMD 指令集（如 AVX2/AVX512）或 JIT 编译技术来加速。

7.4. 生产级框架

本讲座中的 Task<T> 和 ThreadPoolExecutor 是为了演示目的而简化的。在生产环境中，你会希望使用更成熟、功能更丰富的异步框架，例如：

• Boost.ASIO: 提供了强大的异步 I/O 能力和协程支持 (boost::asio::co_spawn)。
• liburing: Linux 上的高性能异步 I/O 接口。
• 自定义事件循环/调度器: 根据应用场景定制。

7.5. 协程栈管理

C++ 协程的帧默认分配在堆上，这可能导致频繁的堆分配和缓存局部性问题。对于性能敏感的应用，可以考虑：

• 自定义分配器： 为协程帧提供一个内存池。
• _Resumable_size 属性（GCC/Clang 扩展）： 允许控制协程帧的大小，但在标准 C++ 中尚无此功能。

---

8. 总结

通过本次讲座，我们探讨了使用 C++20 协程实现异步 JSON 解析器的核心思想和具体实践。我们看到了协程如何将复杂的异步逻辑转化为看似同步的、易于理解和维护的代码。通过构建 Task<T>、ThreadPoolExecutor、AwaitableBufferReader、AsyncJsonLexer 和 AsyncJsonParser，我们成功地演示了在处理 GB 级 JSON 数据的同时，如何确保主线程的响应性和非阻塞性。

C++ 协程为构建高性能、高并发、响应迅速的应用程序开辟了新的道路，它代表了 C++ 异步编程的未来方向。理解并掌握协程的机制，将使你能够以更优雅和高效的方式应对现代软件开发中的诸多挑战。