C++でキューパターンを使った非同期タスク処理の実装方法

C++での非同期処理は、プログラムのパフォーマンス向上とユーザー体験の改善に大きく寄与します。特に、タスクを効率的に管理するためのキューパターンは、非同期処理の実装において非常に有効です。本記事では、キューパターンの基本概念から始め、C++での具体的な実装方法やエラーハンドリング、応用例などを通じて、非同期タスク処理の基礎から応用までを網羅的に解説します。非同期処理におけるキューパターンの力を理解し、実際のプログラムに活かすための知識を身に付けましょう。

キューパターンとは何か

キューパターンは、複数のタスクを順序よく管理し、効率的に処理するための設計パターンです。このパターンでは、タスクをキュー（待ち行列）に入れ、先入れ先出し（FIFO）方式で処理します。これにより、タスクの順序が保証され、スレッドセーフな非同期処理が実現できます。

基本概念

キューパターンの基本概念は、以下の通りです：

キュー：タスクを格納するデータ構造。通常はFIFO（First-In-First-Out）方式が使われる。
プロデューサー：タスクを生成し、キューに追加する役割。
コンシューマー：キューからタスクを取り出し、実行する役割。

非同期処理における役割

非同期処理において、キューパターンは以下のような役割を果たします：

タスク管理の簡素化：タスクの順序を管理しやすくし、複数のタスクが同時に処理されることを防ぐ。
スレッドセーフティ：キューを介してタスクを管理することで、スレッド間の競合を避け、安全に非同期処理を実行できる。
効率的なリソース利用：スレッドプールなどと組み合わせることで、システムリソースを効率的に利用し、パフォーマンスを向上させる。

このように、キューパターンは非同期処理をシンプルかつ効果的に実現するための重要な設計パターンです。次に、C++での具体的なキューの実装方法を見ていきましょう。

C++でのキューの実装

C++でキューパターンを実装するためには、標準ライブラリのstd::queueを使用するのが一般的です。これにより、タスクを効率的に管理し、順序通りに処理することができます。以下に、基本的なキューの実装例を紹介します。

基本的なキューの実装

まずは、C++の標準ライブラリを使ってキューを実装する基本的な方法を見ていきます。

#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>

std::queue<int> taskQueue;
std::mutex queueMutex;
std::condition_variable cv;

void producer(int id) {
    std::unique_lock<std::mutex> lock(queueMutex);
    taskQueue.push(id);
    std::cout << "Produced task " << id << std::endl;
    cv.notify_one();
}

void consumer() {
    std::unique_lock<std::mutex> lock(queueMutex);
    while (taskQueue.empty()) {
        cv.wait(lock);
    }
    int task = taskQueue.front();
    taskQueue.pop();
    std::cout << "Consumed task " << task << std::endl;
}

キューを使った非同期タスクの追加と処理

上記のコードでは、producer関数が新しいタスクをキューに追加し、consumer関数がキューからタスクを取り出して処理します。この基本的な例では、条件変数cvを使用してキューが空の場合に消費者スレッドを待機させ、タスクが追加されたときに通知します。

プロデューサーの実装

プロデューサーは、タスクを生成してキューに追加する役割を持ちます。以下にプロデューサーの実装例を示します。

void producer(int id) {
    std::unique_lock<std::mutex> lock(queueMutex);
    taskQueue.push(id);
    std::cout << "Produced task " << id << std::endl;
    cv.notify_one();
}

コンシューマーの実装

コンシューマーは、キューからタスクを取り出し、それを処理します。以下にコンシューマーの実装例を示します。

void consumer() {
    std::unique_lock<std::mutex> lock(queueMutex);
    while (taskQueue.empty()) {
        cv.wait(lock);
    }
    int task = taskQueue.front();
    taskQueue.pop();
    std::cout << "Consumed task " << task << std::endl;
}

この基本的なキューの実装により、C++で非同期タスクの管理を簡単に行うことができます。次に、非同期タスクの管理方法とキューパターンを用いた効率的な処理方法について解説します。

非同期タスクの管理

非同期タスクを管理する際には、キューパターンを活用することで、タスクの順序を保持しつつ効率的に処理することが可能です。ここでは、タスクの管理方法とキューパターンを用いた非同期処理の実践的な手法について解説します。

タスク管理の基本

非同期タスクを管理する際の基本的なポイントは以下の通りです：

タスクの追加：タスクをキューに追加するプロセス。
タスクの取り出し：キューからタスクを取り出して処理するプロセス。
スレッドセーフティ：複数のスレッドが同時にタスクを追加・取り出しを行う場合の安全性。

キューパターンによる効率的な処理

キューパターンを使うことで、非同期タスクを効率的に管理できます。以下にその手法を示します。

タスクの追加

タスクをキューに追加するためには、適切なロック機構を使用して、スレッドセーフにタスクを追加する必要があります。以下にその実装例を示します。

void addTask(int task) {
    std::unique_lock<std::mutex> lock(queueMutex);
    taskQueue.push(task);
    std::cout << "Task added: " << task << std::endl;
    cv.notify_one();
}

タスクの取り出しと処理

タスクをキューから取り出して処理する際には、条件変数を利用してキューが空の場合にスレッドを待機させることが重要です。以下にその実装例を示します。

void processTasks() {
    while (true) {
        std::unique_lock<std::mutex> lock(queueMutex);
        while (taskQueue.empty()) {
            cv.wait(lock);
        }
        int task = taskQueue.front();
        taskQueue.pop();
        lock.unlock(); // ロックを解除してからタスクを処理
        std::cout << "Processing task: " << task << std::endl;
        // タスクの処理
    }
}

キューパターンを用いた非同期タスクの全体フロー

以下に、キューパターンを用いた非同期タスク処理の全体フローを示します。

タスク生成：プロデューサーが新しいタスクを生成し、キューに追加する。
タスク待機：キューが空の場合、コンシューマースレッドは条件変数を使って待機する。
タスク取り出し：コンシューマーがキューからタスクを取り出し、処理する。
タスク処理：タスクの具体的な処理を行う。

このフローにより、タスクの順序を保ちながら効率的に非同期処理を実現できます。次に、スレッドプールを活用してキューパターンをさらに最適化する方法について説明します。

スレッドプールの活用

スレッドプールを活用することで、キューパターンをさらに効率的に運用できます。スレッドプールは、複数のスレッドをプールしておき、必要に応じてこれらのスレッドを再利用する仕組みです。これにより、スレッドの生成と破棄にかかるオーバーヘッドを削減し、タスクの処理を迅速かつ効率的に行うことが可能です。

スレッドプールの基本概念

スレッドプールの基本的な概念は以下の通りです：

スレッドプールの作成：あらかじめ一定数のスレッドを作成し、待機させる。
タスクの割り当て：新しいタスクが到着した際に、待機中のスレッドにタスクを割り当てる。
スレッドの再利用：タスク処理後にスレッドを破棄せず、再度待機状態に戻す。

C++でのスレッドプールの実装

以下に、C++でスレッドプールを実装するための基本的なコード例を示します。

#include <iostream>
#include <queue>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>

class ThreadPool {
public:
    ThreadPool(size_t numThreads);
    ~ThreadPool();
    void enqueueTask(std::function<void()> task);

private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    std::mutex queueMutex;
    std::condition_variable cv;
    bool stop;

    void workerThread();
};

ThreadPool::ThreadPool(size_t numThreads) : stop(false) {
    for (size_t i = 0; i < numThreads; ++i) {
        workers.emplace_back([this] { this->workerThread(); });
    }
}

ThreadPool::~ThreadPool() {
    {
        std::unique_lock<std::mutex> lock(queueMutex);
        stop = true;
    }
    cv.notify_all();
    for (std::thread &worker : workers) {
        worker.join();
    }
}

void ThreadPool::enqueueTask(std::function<void()> task) {
    {
        std::unique_lock<std::mutex> lock(queueMutex);
        tasks.push(task);
    }
    cv.notify_one();
}

void ThreadPool::workerThread() {
    while (true) {
        std::function<void()> task;
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            cv.wait(lock, [this] { return this->stop || !this->tasks.empty(); });
            if (this->stop && this->tasks.empty()) return;
            task = std::move(this->tasks.front());
            this->tasks.pop();
        }
        task();
    }
}

スレッドプールの利用方法

スレッドプールを利用する際には、タスクをキューに追加するようにします。以下に、その利用方法を示します。

int main() {
    ThreadPool pool(4); // 4つのスレッドを持つスレッドプールを作成

    for (int i = 0; i < 10; ++i) {
        pool.enqueueTask([i] {
            std::cout << "Processing task " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        });
    }

    std::this_thread::sleep_for(std::chrono::seconds(2)); // タスクが完了するのを待つ
    return 0;
}

この例では、4つのスレッドを持つスレッドプールを作成し、10個のタスクをキューに追加しています。各タスクはスレッドプール内のスレッドによって並行して処理されます。

スレッドプールとキューパターンの組み合わせ

スレッドプールをキューパターンと組み合わせることで、タスクの管理と処理を効率化し、リソースの最適な利用を実現します。キューパターンによるタスクの順序管理と、スレッドプールによる効率的なリソース利用を組み合わせることで、非同期タスクの処理性能が大幅に向上します。

次に、具体的なC++コードを使って非同期タスクキューの実装例を紹介します。

実装例：非同期タスクキュー

ここでは、キューパターンとスレッドプールを組み合わせて非同期タスクキューを実装する具体例を紹介します。この実装例を通して、C++での非同期タスク処理を効果的に行う方法を理解しましょう。

全体の構造

まず、非同期タスクキューの全体的な構造を示します。基本的なクラス構造は以下の通りです：

TaskQueue：タスクキューの管理を行うクラス。
ThreadPool：スレッドプールの管理を行うクラス。

TaskQueueの実装

TaskQueueは、タスクをキューに追加し、スレッドプールのスレッドによって処理されるまでの管理を行います。

#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <vector>

class TaskQueue {
public:
    void enqueueTask(std::function<void()> task);
    std::function<void()> dequeueTask();

private:
    std::queue<std::function<void()>> tasks;
    std::mutex queueMutex;
    std::condition_variable cv;
};

void TaskQueue::enqueueTask(std::function<void()> task) {
    {
        std::unique_lock<std::mutex> lock(queueMutex);
        tasks.push(task);
    }
    cv.notify_one();
}

std::function<void()> TaskQueue::dequeueTask() {
    std::unique_lock<std::mutex> lock(queueMutex);
    cv.wait(lock, [this] { return !tasks.empty(); });
    auto task = tasks.front();
    tasks.pop();
    return task;
}

ThreadPoolの実装

ThreadPoolは、複数のスレッドを管理し、タスクキューからタスクを取得して実行します。

class ThreadPool {
public:
    ThreadPool(size_t numThreads, TaskQueue& taskQueue);
    ~ThreadPool();

private:
    std::vector<std::thread> workers;
    TaskQueue& taskQueue;
    bool stop;

    void workerThread();
};

ThreadPool::ThreadPool(size_t numThreads, TaskQueue& tq) : taskQueue(tq), stop(false) {
    for (size_t i = 0; i < numThreads; ++i) {
        workers.emplace_back([this] { this->workerThread(); });
    }
}

ThreadPool::~ThreadPool() {
    {
        std::unique_lock<std::mutex> lock(taskQueue.queueMutex);
        stop = true;
    }
    taskQueue.cv.notify_all();
    for (std::thread &worker : workers) {
        worker.join();
    }
}

void ThreadPool::workerThread() {
    while (true) {
        auto task = taskQueue.dequeueTask();
        if (stop && !task) return;
        task();
    }
}

メイン関数での利用

これらのクラスを利用して、非同期タスクを処理するメイン関数の例を示します。

int main() {
    TaskQueue taskQueue;
    ThreadPool pool(4, taskQueue); // 4つのスレッドを持つスレッドプールを作成

    for (int i = 0; i < 10; ++i) {
        taskQueue.enqueueTask([i] {
            std::cout << "Processing task " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        });
    }

    std::this_thread::sleep_for(std::chrono::seconds(2)); // タスクが完了するのを待つ
    return 0;
}

この例では、TaskQueueにタスクを追加し、ThreadPoolによってそれらのタスクが並行して処理されます。これにより、タスクの順序を保ちながら、効率的に非同期処理を実行することができます。

次に、非同期処理におけるエラーハンドリングの方法とベストプラクティスについて解説します。

エラーハンドリング

非同期処理においては、エラーハンドリングが非常に重要です。タスクが失敗した場合でもシステム全体が停止しないように、適切なエラーハンドリングを実装する必要があります。ここでは、非同期タスクにおけるエラーハンドリングの方法とベストプラクティスについて解説します。

エラーハンドリングの基本

エラーハンドリングの基本的なアプローチは以下の通りです：

例外のキャッチ：タスク内で発生する例外をキャッチして適切に処理する。
エラーログの記録：エラーが発生した際に詳細なログを記録する。
リトライ機構：一部のエラーに対しては、タスクを再試行する機能を持たせる。

タスク内での例外処理

各タスク内で発生する例外をキャッチし、適切に処理することで、システム全体への影響を最小限に抑えることができます。以下に例を示します。

void exampleTask(int id) {
    try {
        // タスクの処理内容
        if (id % 2 == 0) {
            throw std::runtime_error("Example error");
        }
        std::cout << "Processing task " << id << std::endl;
    } catch (const std::exception &e) {
        std::cerr << "Error in task " << id << ": " << e.what() << std::endl;
        // エラーログを記録するなどの追加処理
    }
}

スレッドプール内での例外処理

スレッドプールのワーカースレッド内で例外を処理する方法を示します。各タスクの例外をキャッチし、ログを記録することが重要です。

void ThreadPool::workerThread() {
    while (true) {
        auto task = taskQueue.dequeueTask();
        if (stop && !task) return;
        try {
            task();
        } catch (const std::exception &e) {
            std::cerr << "Worker thread caught an exception: " << e.what() << std::endl;
            // 追加のエラーハンドリング（例：リトライ機構）
        }
    }
}

リトライ機構の実装

特定のエラーに対してタスクを再試行するリトライ機構を実装することも有効です。以下に、リトライ機構の基本的な例を示します。

void exampleTaskWithRetry(int id, int maxRetries) {
    int retryCount = 0;
    while (retryCount < maxRetries) {
        try {
            // タスクの処理内容
            if (id % 2 == 0) {
                throw std::runtime_error("Example error");
            }
            std::cout << "Processing task " << id << std::endl;
            break; // 成功した場合はループを抜ける
        } catch (const std::exception &e) {
            std::cerr << "Error in task " << id << ": " << e.what() << std::endl;
            retryCount++;
            if (retryCount == maxRetries) {
                std::cerr << "Task " << id << " failed after " << maxRetries << " attempts." << std::endl;
            } else {
                std::cerr << "Retrying task " << id << " (" << retryCount << "/" << maxRetries << ")" << std::endl;
            }
            // エラーログを記録するなどの追加処理
        }
    }
}

ベストプラクティス

非同期タスクのエラーハンドリングにおけるベストプラクティスをいくつか挙げます：

一貫したエラーハンドリング：すべてのタスクで一貫したエラーハンドリングの実装を行う。
詳細なエラーログ：エラーが発生した際に、詳細なログを記録し、後から問題を追跡できるようにする。
ユーザーフィードバック：必要に応じて、エラーが発生した際にユーザーにフィードバックを提供する。

次に、非同期処理のテスト方法とデバッグのコツについて紹介します。

テストとデバッグ

非同期処理のテストとデバッグは、同期処理と比べて難易度が高くなります。スレッドの競合状態やタイミングの問題が発生しやすいため、注意深く行う必要があります。ここでは、非同期処理のテスト方法とデバッグのコツについて紹介します。

テストの基本戦略

非同期処理のテストにおいて重要な戦略は以下の通りです：

ユニットテスト：個々のコンポーネントや関数を独立してテストする。
統合テスト：複数のコンポーネントが連携して動作するかをテストする。
負荷テスト：高負荷下でのシステムの動作を確認する。
エラーハンドリングのテスト：意図的にエラーを発生させ、適切に処理されるかを確認する。

ユニットテスト

ユニットテストは、個々の関数やメソッドの正確な動作を確認するためのテストです。以下に、C++でユニットテストを行うための基本的な方法を示します。

#include <gtest/gtest.h>
#include <queue>
#include <functional>

TEST(TaskQueueTest, EnqueueAndDequeue) {
    TaskQueue taskQueue;
    taskQueue.enqueueTask([]{ std::cout << "Task 1" << std::endl; });
    taskQueue.enqueueTask([]{ std::cout << "Task 2" << std::endl; });

    auto task1 = taskQueue.dequeueTask();
    auto task2 = taskQueue.dequeueTask();

    ASSERT_NE(task1, nullptr);
    ASSERT_NE(task2, nullptr);
}

統合テスト

統合テストでは、異なるコンポーネントが正しく連携して動作するかを確認します。以下に、スレッドプールとタスクキューを統合してテストする例を示します。

TEST(ThreadPoolTest, ProcessTasks) {
    TaskQueue taskQueue;
    ThreadPool pool(4, taskQueue);

    for (int i = 0; i < 10; ++i) {
        taskQueue.enqueueTask([i] {
            std::cout << "Processing task " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        });
    }

    std::this_thread::sleep_for(std::chrono::seconds(2)); // タスクが完了するのを待つ
    ASSERT_TRUE(taskQueue.isEmpty());
}

負荷テスト

負荷テストでは、システムが高負荷下でどのように動作するかを確認します。以下に、負荷テストの例を示します。

TEST(LoadTest, HighVolumeTasks) {
    TaskQueue taskQueue;
    ThreadPool pool(10, taskQueue); // 10スレッドのプール

    for (int i = 0; i < 1000; ++i) {
        taskQueue.enqueueTask([i] {
            std::cout << "Processing high volume task " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        });
    }

    std::this_thread::sleep_for(std::chrono::seconds(10)); // タスクが完了するのを待つ
    ASSERT_TRUE(taskQueue.isEmpty());
}

エラーハンドリングのテスト

エラーハンドリングのテストでは、意図的にエラーを発生させて、そのエラーが適切に処理されるかを確認します。

TEST(ErrorHandlingTest, TaskError) {
    TaskQueue taskQueue;
    ThreadPool pool(4, taskQueue);

    taskQueue.enqueueTask([] {
        throw std::runtime_error("Test error");
    });

    std::this_thread::sleep_for(std::chrono::seconds(1)); // タスクが完了するのを待つ
    // エラーが発生したことを確認
}

デバッグのコツ

非同期処理のデバッグは難しいですが、以下のコツを押さえておくと効果的です：

ログの活用：各スレッドやタスクの状態を詳細にログに記録し、問題の発生箇所を特定する。
デバッガの使用：IDEのデバッガを活用し、スレッドの状態や変数の値をリアルタイムで確認する。
シンプルな再現ケース：問題を簡単に再現できる最小限のコードを作成し、そこでデバッグを行う。

次に、キューパターンを用いたリアルタイムシステムでの非同期処理の応用例について説明します。

応用例：リアルタイムシステム

キューパターンを用いた非同期処理は、リアルタイムシステムにおいても非常に有効です。リアルタイムシステムでは、迅速かつ正確な処理が求められるため、効率的なタスク管理が不可欠です。ここでは、リアルタイムシステムにおけるキューパターンの応用例を紹介します。

リアルタイムシステムとは

リアルタイムシステムは、外部からの入力に対して決められた時間内に応答する必要があるシステムです。具体的な例としては、以下のようなシステムがあります：

工業用制御システム：工場の生産ラインを管理するシステム。
航空制御システム：航空機の運行を管理するシステム。
自動運転車システム：自動車の運行を制御するシステム。

キューパターンの適用例

リアルタイムシステムでのキューパターンの具体的な適用例を以下に示します。

工業用制御システム

工業用制御システムでは、センサーからのデータをリアルタイムで処理し、適切な制御信号を出力する必要があります。キューパターンを用いることで、センサーデータを効率的に管理し、スレッドプールを利用して並行処理を行うことができます。

#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <functional>

class RealTimeTaskQueue {
public:
    void enqueueTask(std::function<void()> task);
    std::function<void()> dequeueTask();

private:
    std::queue<std::function<void()>> tasks;
    std::mutex queueMutex;
    std::condition_variable cv;
};

void RealTimeTaskQueue::enqueueTask(std::function<void()> task) {
    {
        std::unique_lock<std::mutex> lock(queueMutex);
        tasks.push(task);
    }
    cv.notify_one();
}

std::function<void()> RealTimeTaskQueue::dequeueTask() {
    std::unique_lock<std::mutex> lock(queueMutex);
    cv.wait(lock, [this] { return !tasks.empty(); });
    auto task = tasks.front();
    tasks.pop();
    return task;
}

class RealTimeThreadPool {
public:
    RealTimeThreadPool(size_t numThreads, RealTimeTaskQueue& taskQueue);
    ~RealTimeThreadPool();

private:
    std::vector<std::thread> workers;
    RealTimeTaskQueue& taskQueue;
    bool stop;

    void workerThread();
};

RealTimeThreadPool::RealTimeThreadPool(size_t numThreads, RealTimeTaskQueue& tq) : taskQueue(tq), stop(false) {
    for (size_t i = 0; i < numThreads; ++i) {
        workers.emplace_back([this] { this->workerThread(); });
    }
}

RealTimeThreadPool::~RealTimeThreadPool() {
    {
        std::unique_lock<std::mutex> lock(taskQueue.queueMutex);
        stop = true;
    }
    taskQueue.cv.notify_all();
    for (std::thread &worker : workers) {
        worker.join();
    }
}

void RealTimeThreadPool::workerThread() {
    while (true) {
        auto task = taskQueue.dequeueTask();
        if (stop && !task) return;
        try {
            task();
        } catch (const std::exception &e) {
            std::cerr << "Worker thread caught an exception: " << e.what() << std::endl;
        }
    }
}

int main() {
    RealTimeTaskQueue taskQueue;
    RealTimeThreadPool pool(4, taskQueue);

    for (int i = 0; i < 10; ++i) {
        taskQueue.enqueueTask([i] {
            std::cout << "Processing real-time task " << i << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        });
    }

    std::this_thread::sleep_for(std::chrono::seconds(2));
    return 0;
}

自動運転車システムの例

自動運転車では、センサーからのデータをリアルタイムで処理し、車両の制御に反映する必要があります。キューパターンを利用することで、センサーデータの処理と制御命令の発行を効率的に行うことができます。

class AutonomousVehicleSystem {
public:
    void processSensorData();
    void controlVehicle();

private:
    RealTimeTaskQueue taskQueue;
    RealTimeThreadPool threadPool;

    AutonomousVehicleSystem() : threadPool(4, taskQueue) {}

    void sensorDataTask();
    void controlTask();
};

void AutonomousVehicleSystem::processSensorData() {
    taskQueue.enqueueTask([this] { sensorDataTask(); });
}

void AutonomousVehicleSystem::controlVehicle() {
    taskQueue.enqueueTask([this] { controlTask(); });
}

void AutonomousVehicleSystem::sensorDataTask() {
    // センサーからデータを取得して処理する
    std::cout << "Processing sensor data" << std::endl;
}

void AutonomousVehicleSystem::controlTask() {
    // 車両を制御する
    std::cout << "Controlling vehicle" << std::endl;
}

int main() {
    AutonomousVehicleSystem avs;

    // シミュレーションとして連続的にセンサーデータを処理し、車両を制御する
    for (int i = 0; i < 10; ++i) {
        avs.processSensorData();
        avs.controlVehicle();
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }

    std::this_thread::sleep_for(std::chrono::seconds(5));
    return 0;
}

このように、キューパターンを用いることで、リアルタイムシステムにおける非同期タスク処理が効率的かつ安全に実現できます。次に、理解を深めるための演習問題を提供します。

演習問題

ここでは、非同期処理とキューパターンに関する理解を深めるための演習問題を提供します。これらの問題を通じて、実際にコードを書きながら学びを深めてください。

演習問題 1: 基本的なキューの実装

以下の要件に従って、基本的なタスクキューを実装してください。

タスクをキューに追加するenqueueTaskメソッドを実装する。
タスクをキューから取り出すdequeueTaskメソッドを実装する。
複数のスレッドからタスクを追加し、キューから取り出して処理するコードを書く。

#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>

class TaskQueue {
public:
    void enqueueTask(std::function<void()> task);
    std::function<void()> dequeueTask();

private:
    std::queue<std::function<void()>> tasks;
    std::mutex queueMutex;
    std::condition_variable cv;
};

void TaskQueue::enqueueTask(std::function<void()> task) {
    // ここにコードを追加してください
}

std::function<void()> TaskQueue::dequeueTask() {
    // ここにコードを追加してください
}

void producer(TaskQueue& taskQueue, int id) {
    taskQueue.enqueueTask([id] {
        std::cout << "Produced task " << id << std::endl;
    });
}

void consumer(TaskQueue& taskQueue) {
    while (true) {
        auto task = taskQueue.dequeueTask();
        task();
    }
}

int main() {
    TaskQueue taskQueue;

    std::thread producerThread(producer, std::ref(taskQueue), 1);
    std::thread consumerThread(consumer, std::ref(taskQueue));

    producerThread.join();
    consumerThread.join();

    return 0;
}

演習問題 2: スレッドプールの実装

スレッドプールを実装し、タスクキューと連携させてタスクを並行処理するコードを書いてください。

スレッドプールクラスThreadPoolを実装する。
タスクをスレッドプール内のスレッドで並行処理するworkerThreadメソッドを実装する。

#include <iostream>
#include <queue>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>

class TaskQueue {
public:
    void enqueueTask(std::function<void()> task);
    std::function<void()> dequeueTask();

private:
    std::queue<std::function<void()>> tasks;
    std::mutex queueMutex;
    std::condition_variable cv;
};

class ThreadPool {
public:
    ThreadPool(size_t numThreads, TaskQueue& taskQueue);
    ~ThreadPool();

private:
    std::vector<std::thread> workers;
    TaskQueue& taskQueue;
    bool stop;

    void workerThread();
};

void TaskQueue::enqueueTask(std::function<void()> task) {
    // ここにコードを追加してください
}

std::function<void()> TaskQueue::dequeueTask() {
    // ここにコードを追加してください
}

ThreadPool::ThreadPool(size_t numThreads, TaskQueue& tq) : taskQueue(tq), stop(false) {
    // ここにコードを追加してください
}

ThreadPool::~ThreadPool() {
    // ここにコードを追加してください
}

void ThreadPool::workerThread() {
    // ここにコードを追加してください
}

int main() {
    TaskQueue taskQueue;
    ThreadPool pool(4, taskQueue);

    for (int i = 0; i < 10; ++i) {
        taskQueue.enqueueTask([i] {
            std::cout << "Processing task " << i << std::endl;
        });
    }

    std::this_thread::sleep_for(std::chrono::seconds(2));
    return 0;
}

演習問題 3: エラーハンドリングの追加

非同期タスクに対するエラーハンドリングを実装してください。

タスク内で例外が発生した場合に、その例外をキャッチしてログに記録する。
workerThreadメソッド内で例外を処理し、システム全体の動作が停止しないようにする。

#include <iostream>
#include <queue>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <stdexcept>

class TaskQueue {
public:
    void enqueueTask(std::function<void()> task);
    std::function<void()> dequeueTask();

private:
    std::queue<std::function<void()>> tasks;
    std::mutex queueMutex;
    std::condition_variable cv;
};

class ThreadPool {
public:
    ThreadPool(size_t numThreads, TaskQueue& taskQueue);
    ~ThreadPool();

private:
    std::vector<std::thread> workers;
    TaskQueue& taskQueue;
    bool stop;

    void workerThread();
};

void TaskQueue::enqueueTask(std::function<void()> task) {
    // ここにコードを追加してください
}

std::function<void()> TaskQueue::dequeueTask() {
    // ここにコードを追加してください
}

ThreadPool::ThreadPool(size_t numThreads, TaskQueue& tq) : taskQueue(tq), stop(false) {
    // ここにコードを追加してください
}

ThreadPool::~ThreadPool() {
    // ここにコードを追加してください
}

void ThreadPool::workerThread() {
    while (true) {
        auto task = taskQueue.dequeueTask();
        if (stop && !task) return;
        try {
            task();
        } catch (const std::exception &e) {
            std::cerr << "Worker thread caught an exception: " << e.what() << std::endl;
        }
    }
}

int main() {
    TaskQueue taskQueue;
    ThreadPool pool(4, taskQueue);

    for (int i = 0; i < 10; ++i) {
        taskQueue.enqueueTask([i] {
            if (i % 2 == 0) {
                throw std::runtime_error("Test error");
            }
            std::cout << "Processing task " << i << std::endl;
        });
    }

    std::this_thread::sleep_for(std::chrono::seconds(2));
    return 0;
}

これらの演習問題を通して、非同期タスクの管理、スレッドプールの実装、エラーハンドリングの技術を身につけてください。次に、この記事のまとめを行います。

まとめ

この記事では、C++でキューパターンを使った非同期タスク処理の基礎から応用までを詳細に解説しました。まず、キューパターンの基本概念とその利点について説明し、次にC++での具体的なキューの実装方法を紹介しました。さらに、非同期タスクの管理方法やスレッドプールの活用方法についても詳述しました。

キューパターンは、タスクの順序を管理し、効率的に非同期処理を実現するための強力な設計パターンです。スレッドプールを併用することで、システムのパフォーマンスを大幅に向上させることができます。また、エラーハンドリングやテストの重要性についても強調しました。これにより、信頼性の高い非同期システムを構築するための基本的な知識と技術を身につけることができたでしょう。

演習問題を通して、実際のコードを書くことで理解を深めることができたと思います。この記事の内容を実際のプロジェクトに活用し、より効率的で信頼性の高い非同期タスク処理を実現してください。