POCO::Foundationでデザインパターン - マルチスレッド編 -

オープンソースC++用クラスライブラリPOCO活用講座（5）

POCOによるワーカースレッド（Worker Thread）・パターン

　スレッド関連クラスを応用して、ワーカースレッド・パターンのサンプルプログラムを作ってみました。POCOのソースと一緒に配布されるサンプルコードを土台として、若干の変更を加えました。

　ワーカースレッド・パターンの動きをざっと説明すると、次のようになります。

1. まず複数のワーカースレッド（作業スレッド）を生成します。
2. 各スレッドは、共有のリソースに対して「仕事」をとりにいきます。
仕事がある場合はその作業を行い、ない場合は新しい仕事が見つかるまで待機します。
3. 共有のリソースに仕事を追加していきます。
4. 順次ワーカースレッドが手分けして処理を行います。
5. 仕事の追加を終えたら、すべての仕事が処理されるまで待機します。

サンプルコード

　まず最初にサンプルコードを紹介することにします。

ワーカースレッド・パターン　サンプルコード

#include "Poco/Notification.h"
#include "Poco/NotificationQueue.h"
#include "Poco/ThreadPool.h"
#include "Poco/Thread.h"
#include "Poco/Runnable.h"
#include "Poco/Mutex.h"
#include "Poco/Random.h"

#include "Poco/AutoPtr.h"

#include <iostream>

// ワーカースレッドへの通知クラス
class WorkNotification: public Poco::Notification
{
  // 通知の番号
  int    _ix;

public:
  WorkNotification( int i ): _ix(i)
  {
  }

  // 保持している番号を返す
  int ix() const
  {
    return _ix;
  }

  // コマンド処理
  void execute()
  {
    // 乱数クラス
    Poco::Random  rnd;

    // 同じ乱数列とならないように
    rnd.seed();

    // 乱数でSleepする（0～199ミリ秒）
    Poco::Thread::sleep( rnd.next(200) );
  }
};

// ワーカースレッドクラス
// 通知キューから通知オブジェクトを取り出し、実行する。
class Worker: public Poco::Runnable
{
  // スレッド識別名
  std::string         _name;

  // 通知キュー
  Poco::NotificationQueue&  _queue;

  // 排他処理用Mutex
  static Poco::FastMutex    _mutex;

public:
  Worker(const std::string& name, Poco::NotificationQueue& queue):
    _name(name), _queue(queue)
  {
  }

  // スレッドのメイン処理
  void run()
  {
    for (;;) {
      // 通知キューからデータが取得できるまで待機
      Poco::AutoPtr<Poco::Notification> pNf(_queue.waitDequeueNotification());
      if (pNf) {
        WorkNotification* pWorkNf = dynamic_cast<WorkNotification*>(pNf.get());
        if (pWorkNf) {
          {
            Poco::FastMutex::ScopedLock lock(_mutex);
            std::cout << _name << " excute No." << pWorkNf->ix() << std::endl;
          }
          // コマンド実行
          pWorkNf->execute();
        }
      }
      else
        break;  // 返値がNULLのときはスレッドを終了
    }
  }
};
Poco::FastMutex Worker::_mutex; // static変数の実体

int _tmain(int argc, _TCHAR* argv[])
{
  // 通知オブジェクトのキュー
  Poco::NotificationQueue queue;

  // ワーカースレッドを作成
  Worker worker1( "Worker-1", queue );
  Worker worker2( "Worker-2", queue );
  Worker worker3( "Worker-3", queue );

  // スレッドプール
  Poco::ThreadPool pool;

  // 各ワーカースレッドの登録と開始
  pool.start( worker1 );
  pool.start( worker2 );
  pool.start( worker3 );

  // 通知オブジェクト50個をキューに積む
  for (int i = 0; i < 50; ++i) {
    queue.enqueueNotification( new WorkNotification( i ) );
                         // 順番を渡す
  }

  // キューが空になるまで、メインスレッドは待機。
  while ( !queue.empty() ) Poco::Thread::sleep(200);

  // すべてのワーカースレッドが処理待ちになるまでSleepしておく
  Poco::Thread::sleep(200);

  // 全ワーカースレッドへ終了指示
  queue.wakeUpAll();

  // 全ワーカースレッドの終了待機
  pool.joinAll();

  return 0;
}

　このサンプルコードに新たに登場するPOCOのクラスは次のとおりです。

その他のクラス

クラス名	概要
Notification	通知クラス
NotificationQueue	通知クラス用キュー構造のコンテナクラス
Poco::Random	乱数を生成するクラス

　POCOでは複数のスレッドを扱うためにPoco::ThreadPool（スレッドプール）というクラスが提供されています。スレッドの生成はそれなりのコストがかかる処理ですから、生成と廃棄を繰り返すことは効率的ではありません。そこでThreadPoolクラスでは、あらかじめスレッドを生成し、それをプールしておきます。必要に応じてプールしてあるスレッドを使いまわすことで、効率のよい複数スレッドの管理を実現できるようになっています。ちなみに、Javaや.NETのクラスライブラリにも、同じ名前で同様のクラスが存在します。

　前述の説明で用いた「仕事」という概念は、POCOではNotification（通知）という考え方で実装しています。ワーカースレッドに対して、メッセージなどを「通知」するということです。通知の受け渡しを実装したクラスとして、Poco::Notificationクラス、Poco::NotificationQueueクラスが用意されています。NotificationQueueクラスは、Notificationオブジェクトを保有するキュー構造のコンテナクラスです。

　サンプルコードでは単なる通知という役割だけでなく、「仕事」の具体的な処理も行わせるようにクラスを拡張しました。なお、このサンプルコードのように処理をオブジェクトとして捉えたパターンを、コマンドパターンと呼びます。

　POCOによるスレッドプールの概要を図にしてみました。

スレッドプール概要図