(57)【要約】【課題】分散環境上でレコード処理装置が自律的に動作している環境では、各レコード処理装置側にＰｕｓｈモードで動作するかＰｕｌｌモードで動作するかの選択権があり、レコード蓄積装置側からＰｕｓｈモード／Ｐｕｌｌモードで動作せよという指示を出して集中管理することができない。【解決手段】本発明は、レコード処理装置からのモード変更の契機でその時の状況（ＰｕｓｈモードとＰｕｌｌモードで動作するレコード処理装置の割合と第1のキューの蓄積された未処理のレコードの割合）からモード変更を受け入れるかどうかを判断することによって系全体で効率の良い分散データ処理ができるようにレコード蓄積装置が緩やかに系全体を管理する。【選択図】図１
2010117942.tif 000002
【特許請求の範囲】
【請求項１】
少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードを第１のキューに蓄積するレコード蓄積装置と、該レコード蓄積装置からレコードを受信して処理を行うレコード処理装置とが分散配置されており、
少なくとも１つ以上のレコード処理装置が、他のレコード処理装置と何らかの情報共有または情報交換をすることなく、独立かつ並行に前記レコード蓄積装置からレコードを受信して処理を行う分散データ処理システムであって、
前記レコード蓄積装置は、
第１のキューと、
前記クライアントから受信したレコードにレコード識別情報を付加してレコードの到着順（ＦＩＦＯ）に前記第１のキューに蓄積する第１のレコード蓄積手段と、
前記第１のキューから蓄積したレコードを、レコードが到着した順（ＦＩＦＯ）に取り出す第１のレコード抽出手段と、
取り出したレコードを前記レコード処理装置に送信するレコード配信手段と、
を具備し、
前記レコード処理装置は、
第２のキューと、
前記レコード蓄積装置からレコードを取得するレコード取得手段と、
取得した前記レコードを前記第２のキューに到着順（ＦＩＦＯ）に蓄積する第２のレコード蓄積手段と、
前記第２のキューから蓄積したレコードをレコードの到着順（ＦＩＦＯ）に取り出す第２のレコード抽出手段と、
取得したレコードの前記レコード識別情報に応じて、予め決められた処理を実行するレコード処理手段と、
前記第２のレコード抽出手段によって取り出したレコードを前記レコード処理手段によって処理し、レコード処理完了後に、前記第２のレコード抽出手段によって前記第２のキューからレコードを取り出す処理を繰り返す手段を含む
ことを特徴とする分散データ処理システム。
【請求項２】
前記レコード処理装置は、
前記レコード蓄積装置からレコードを受信する前に、該レコード蓄積装置の前記レコード配信手段から配信されたレコードを取得するＰｕｓｈモード、または、前記レコード取得手段により自律的にレコードを取得するＰｕｌｌモードのいずれか一方により、レコードを受信することを該レコード蓄積装置に通知するモード通知手段を有し、
前記レコード蓄積装置は、
前記レコード配信手段によりレコードを受信するレコード処理装置の識別子を記録するＰｕｓｈモードメンバリスト記憶手段と、
前記レコード取得手段によりレコードを受信するレコード処理装置の識別子を記録するＰｕｌｌモードメンバリスト記憶手段と、
前記レコード処理装置からのモードの通知に応じて、前記Ｐｕｓｈモードメンバリスト記憶手段の登録状況と、前記Ｐｕｌｌモードメンバリスト記憶手段の登録状況と、を更新するレコード受信モード制御手段と、
少なくとも１つ以上のレコード処理装置が、前記レコード配信手段または前記レコード取得手段のいずれか一方よりレコード蓄積装置からレコードを受信する際に、
前記Ｐｕｓｈモードの場合には、前記Ｐｕｓｈモードメンバリスト記憶手段に含まれるレコード処理装置の中から何らかの方法で１つのレコード処理装置を選択し、選択したレコード処理装置の第２のキューに空きがあるかどうかを確認し、空きがあれば前記第１のレコード抽出手段によって前記第１のキューから取り出したレコードを配信し、前記Ｐｕｓｈモードメンバリスト記憶手段に含まれる次のレコード処理装置に対して、同様の手順でレコードを配信する処理を繰り返す手段と、
前記Ｐｕｌｌモードの場合には、前記レコード処理手段によるレコード処理完了後（当該レコード処理装置の第２のキューに処理待ちレコードが存在しなくなった時）に、当該レコード処理装置からレコード取得要求を受信し、該レコード処理装置がＰｕｌｌモードメンバリスト記憶手段に含まれている場合に限り、前記第１のレコード抽出手段によって前記第１のキューから取り出されたレコードを送信する処理を繰り返す手段と、
を含む請求項１記載の分散データ処理システム。
【請求項３】
前記レコード処理装置は、
前記Ｐｕｓｈモードでレコードを受信しているとき、前記Ｐｕｌｌモードへ変更する第１のモード変更通知を、レコード蓄積装置に通知する第１のモード変更通知手段と、
前記Ｐｕｌｌモードでレコードを受信しているとき、前記レコード処理手段によるレコード処理完了後に、前記Ｐｕｓｈモードへ変更する第２のモード変更通知を該レコード蓄積装置に通知する第２のモード変更通知手段と、を有し、
前記レコード蓄積装置の前記レコード受信モード制御手段は、
前記第１のモード変更通知を受信すると、前記Ｐｕｓｈモードメンバリスト記憶手段から通知元のレコード処理装置のエントリを削除し、前記Ｐｕｌｌモードメンバリスト記憶手段の末尾に当該レコード処理装置のエントリを追加する手段と、
前記第２のモード変更通知を受信すると、前記Ｐｕｌｌモードメンバリスト記憶手段から通知元のレコード処理装置のエントリを削除し、前記Ｐｕｓｈモードメンバリスト記憶手段の末尾に当該レコード処理装置のエントリを追加する手段と、
を含む請求項２記載の分散データ処理システム。
【請求項４】
前記レコード受信モード制御手段は、
前記第１のモード変更通知または前記第２のモード変更通知を取得すると、前記Ｐｕｓｈモードメンバリスト記憶手段に登録されているレコード処理装置の数と、前記Ｐｕｌｌモードメンバリスト記憶手段に登録されているレコード処理装置の数と、前記第１のキューに蓄積されている処理待ちレコードの数と、該第１のキューに蓄積可能な最大レコードの数からレコード受信モードの変更可否を決定する手段と、
前記レコード配信手段（Ｐｕｓｈモード）によってレコードを受信するレコード処理装置と、前記レコード取得手段（Ｐｕｌｌモード）によってレコードを受信するレコード処理装置の割合を動的に制御する手段と、
を含む請求項２または３記載の分散データ処理システム。
【請求項５】
少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードを第１のキューに蓄積するレコード蓄積装置と、該レコード蓄積装置からレコードを受信して処理を行うレコード処理装置とが分散配置されており、少なくとも１つ以上のレコード処理装置が他のレコード処理装置と何らかの情報共有または情報交換をすることなく、独立かつ並行に前記レコード蓄積装置からレコードを受信して処理を行う分散データ処理システムにおける分散データ処理方法であって、
前記レコード蓄積装置は、
前記クライアントから受信したレコードにレコード識別情報を付加してレコードの到着順（ＦＩＦＯ）に前記第１のキューに蓄積し、該第１のキューから蓄積したレコードを、レコードが到着した順（ＦＩＦＯ）に取り出し、取り出したレコードを前記レコード処理装置に送信し、
前記レコード処理装置は、
前記レコード蓄積装置からレコードを取得し、受信したレコードを到着順（ＦＩＦＯ）に第２のキューに蓄積し、該第２のキューから蓄積したレコードをレコードの到着順（ＦＩＦＯ）に取り出し、レコードの前記レコード識別情報に応じて、予め決められた処理を実行し、レコード処理完了後に、前記第２のキューからレコードを取り出す処理を繰り返す
ことを特徴とする分散データ処理方法。
【請求項６】
前記レコード処理装置は、
前記レコード蓄積装置からレコードを受信する前に、該レコード蓄積装置のレコード配信手段からレコードを受信するＰｕｓｈモード、または、当該レコード処理装置のレコード取得手段により自律的に該レコード蓄積装置からレコードを受信するＰｕｌｌモードのいずれか一方により、レコードを受信することを該レコード蓄積装置に通知し、
前記レコード蓄積装置は、
Ｐｕｓｈモードである場合に、Ｐｕｓｈモードによりレコードを受信するレコード処理装置の識別子を記録するＰｕｓｈモードメンバリスト記憶手段に含まれるレコード処理装置の中から何らかの方法で１つのレコード処理装置を選択し、選択したレコード処理装置の第２のキューに空きがあるかどうかを確認し、空きがあれば前記第１のキューから取り出したレコードを配信し、前記Ｐｕｓｈモードメンバリスト記憶手段に含まれる次のレコード処理装置に対して、同様の手順でレコードを配信する処理を繰り返し、
Ｐｕｌｌモードである場合に、該レコード処理装置がレコード処理完了後（当該レコード処理装置の第２のキューに処理待ちレコードが存在しなくなった時）に、該レコード処理装置からレコードの取得要求があると、Ｐｕｌｌモードによりレコードを受信するレコード処理装置の識別子を記録するＰｕｌｌモードメンバリスト記憶手段に当該レコード処理装置が含まれている場合に限り、前記第１のキューからレコードを取り出し、取り出したレコードを該レコード処理装置に送信する処理を繰り返す、
請求項５記載の分散データ処理方法。
【請求項７】
前記レコード処理装置が、前記Ｐｕｓｈモードでレコードを受信しているとき、前記Ｐｕｌｌモードへ変更する第１のモード変更通知を、レコード蓄積装置に通知し、
前記レコード蓄積装置が、前記第１のモード変更通知を受信すると、前記Ｐｕｓｈモードメンバリスト記憶手段から通知元のレコード処理装置のエントリを削除し、前記Ｐｕｌｌモードメンバリスト記憶手段の末尾に当該レコード処理装置のエントリを追加し、
また、前記レコード処理装置が、前記Ｐｕｌｌモードでレコードを受信しているとき、前記レコード処理手段によるレコード処理完了後に、前記Ｐｕｓｈモードへ変更する第２のモード変更通知を該レコード蓄積装置に通知し、
前記レコード蓄積装置が、前記第２のモード変更通知を受信すると、前記Ｐｕｌｌモードメンバリスト記憶手段から通知元のレコード処理装置のエントリを削除し、前記Ｐｕｓｈモードメンバリスト記憶手段の末尾に当該レコード処理装置のエントリを追加する、
請求項５または６記載の分散データ処理方法。
【請求項８】
前記レコード蓄積装置において、
前記レコード処理装置からＰｕｓｈモードからＰｕｌｌモードへの変更の通知またはＰｕｌｌモードからＰｕｓｈモードへの変更の通知を取得すると、前記Ｐｕｓｈモードメンバリスト記憶手段に登録されているレコード処理装置の数と、前記Ｐｕｌｌモードメンバリスト記憶手段に登録されているレコード処理装置の数と、前記第１のキューに蓄積されている処理待ちレコードの数と、該第１のキューに蓄積可能な最大レコードの数からレコード受信モードの変更可否を決定する手段と、
前記Ｐｕｓｈモードによってレコードを受信するレコード処理装置と、前記Ｐｕｌｌモードによってレコードを受信するレコード処理装置の割合を動的に制御する
請求項７記載の分散データ処理方法。
【請求項９】
少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードを第１のキューに蓄積するレコード蓄積装置と、該レコード蓄積装置からレコードを受信して処理を行うレコード処理装置とが分散配置されており、少なくとも１つ以上のレコード処理装置が他のレコード処理装置と何らかの情報共有または情報交換をすることなく、独立かつ並行に前記レコード蓄積装置からレコードを受信して処理を行う分散データ処理システムにおけるレコード蓄積装置として機能するコンピュータに実行させるプログラムであって、
前記クライアントから受信したレコードにレコード識別情報を付加してレコードの到着順（ＦＩＦＯ）に第１のキューに蓄積し、該第１のキューから蓄積したレコードを、レコードが到着した順（ＦＩＦＯ）に取り出し、取り出したレコードを前記レコード処理装置に送信するレコード送信ステップをコンピュータに実行させ、
前記レコード送信ステップは、
前記レコード処理装置から、レコード蓄積装置からレコードが配信されるＰｕｓｈモードまたは、該レコード処理装置において該レコード蓄積装置よりレコードを取得するＰｕｌｌモードのいずれかのモードが指定されると、
前記Ｐｕｓｈモードの場合は、Ｐｕｓｈモードでレコードを受信するレコード処理装置が登録されているＰｕｓｈモードメンバリスト記憶手段を検索して、何らかの方法で１つのレコード処理装置を選択し、選択したレコード処理装置の第２のキューに空きがあるかどうかを確認し、空きがあれば前記第１のキューから取り出したレコードを配信し、前記Ｐｕｓｈモードメンバリスト記憶手段に含まれる次のレコード処理装置に対して、同様の手順でレコードを配信する処理を繰り返すステップと、
前記Ｐｕｌｌモードの場合は、Ｐｕｌｌモードでレコードを受信するレコード処理装置が登録されているＰｕｌｌモードメンバリスト記憶手段を検索して、当該レコード処理装置がＰｕｌｌモードメンバリストに含まれている場合に限り、前記第１のキューからレコードを取り出し、取り出したレコードを送信する処理を繰り返すステップと、
を含むことを特徴とするプログラム。
【請求項１０】
少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードを第１のキューに蓄積するレコード蓄積装置と、該レコード蓄積装置からレコードを受信して処理を行うレコード処理装置とが分散配置されており、少なくとも１つ以上のレコード処理装置が他のレコード処理装置と何らかの情報共有または情報交換をすることなく、独立かつ並行に前記レコード蓄積装置からレコードを受信して処理を行う分散データ処理システムにおける該レコード処理装置として機能するコンピュータに実行させるプログラムであって、
前記レコード蓄積装置からレコードを受信する前に、該レコード蓄積装置からレコードが配信されるＰｕｓｈモードまたは、該レコード処理装置において該レコード蓄積装置よりレコードを取得するＰｕｌｌモードのいずれかのモードにより、レコードを受信することを該レコード蓄積装置に通知するモード通知ステップと、
レコード蓄積装置からレコードを取得して、取得した順（ＦＩＦＯ）に第２のキューに蓄積するレコード取得ステップと、
前記第２のキューから取得したレコードのレコード識別情報に応じて、予め決められた処理を実行するレコード処理ステップと、
をコンピュータに実行させるプログラム。
【発明の詳細な説明】
【技術分野】
【０００１】
本発明は、分散データ処理システム及び方法及びプログラムに係り、特に、大量のセンサデータや各種情報システムのログ情報、Ｗｅｂやブログなどのコンシューマによって生成されるコンテンツ情報、映像や音声などの各種ストリームデータが日々大量に生成される広域分散ネットワーク上で、データ（レコード）を収集・蓄積・管理し、利用者の目的に応じた処理を行う分散データ処理システム及び方法及びプログラムに関する。
【背景技術】
【０００２】
キューにレコードを蓄積するレコード蓄積装置と、キューから取り出したレコードを処理するレコード処理装置とが分散配置されている分散コンピューティング環境において、負荷分散やスケールアウト、可用性の向上を目的とした分散データ処理方式として、レコード蓄積装置を追記部、レコード処理装置をフィルタ部とし、それぞれを分散するプロセス構成で実現する。系内のプロセス構成を管理する情報部が、系内のプロセス構成を記録したプロセス構成情報をマルチキャストで系内の全プロセスに配信することでプロセスの生存状況や新規プロセスの追加状況を確認し、最新のプロセス構成に更新したプロセス構成情報を配信することを定期的に行う。追記部ではクライアントから入力されたレコードに対して管理用のレコードＩＤ及び時刻印を付与した後でキューにレコードを蓄積する。フィルタ部では情報部からのマルチキャストで受信したプロセス構成情報の中から接続可能な追記部の一覧を取得し、ラウンドロビンアルゴリズムなどによって追記部に接続し、キューから情報源ＩＤと共に、レコードを取得（Ｐｕｌｌ）し、情報源ＩＤに応じて決められた処理を行うステップを繰り返す（例えば、非特許文献１参照）。
【０００３】
このように、従来は、レコード蓄積装置が主導的にレコード処理装置にレコードを配信（Ｐｕｓｈ）する方式ではなく、レコード処理装置が自律的にレコード蓄積装置に蓄積されたレコードを取得（Ｐｕｌｌ）する方式を採用することで、レコード処理装置が過負荷になることを防ぎ、系としての安定性向上を目指している。
【０００４】
更に、分散コンピューティング環境における動作を考慮して、レコード処理装置が自らのタイミングで処理の制御（処理の実行、停止、再起動）ができるようにしている。
【非特許文献１】「追記・参照型データ管理システムの設計と評価」赤間他、情報処理学会論文誌,Vol. 49 No.2, 2008年2月、pp.749-764
【発明の開示】
【発明が解決しようとする課題】
【０００５】
しかしながら、上記従来の方式は、レコード蓄積装置とレコード処理装置とが分散配置された分散コンピューティング環境において、系内の全てのレコード処理装置が主導権を持ってレコード蓄積装置に接続してレコードを取得（Ｐｕｌｌ）し、取得したレコードに対する処理を行い、処理が完了した後にレコード蓄積装置に接続して次に処理すべきレコードを取得するステップを逐次的に繰り返すため、レコードの取得に長い時間がかかる場合やレコードの処理にかかる時間が短い場合（つまり、レコード処理装置がレコード蓄積装置からレコードを取得して処理完了するまでにかかる時間のうち、レコードの取得にかかる時間の割合が無視できない大きさの場合）は、図９（Ｂ）に示すようにキューに蓄積された各レコードの処理完了までに係る時間が長くなり、クライアントからレコードが入力されてから処理結果を出力するまでの遅延が大きくなるという問題点がある。
【０００６】
また、上記従来の方式を、全てのレコード処理装置がレコード蓄積装置からの配信（Ｐｕｓｈ）によってレコードを受信する方式（図９（Ａ））に改良したり、レコード蓄積装置からの配信（Ｐｕｓｈ）によってレコードを受信するレコード処理装置と自らのレコード蓄積装置からレコードを取得（Ｐｕｌｌ）するレコード処理装置とが予め設定した固定の割合で混在する方式（図９（Ｃ））に改良したとしても分散コンピューティング環境上で各レコード処理装置が自律的に動作する特性を保持したままでクライアントから不規則な間隔で入力されるレコードに対して、レコードの入力から処理結果出力までの遅延を小さくすることは困難である。
【０００７】
本発明は、上記の点に鑑みなされたもので、キューにレコードを蓄積するレコード蓄積装置と、キューから取り出したレコードを処理するレコード処理装置とが分散配置されている分散コンピューティング環境において、少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードをキューに蓄積し、蓄積されたレコードを取得して利用者の目的に応じた処理を行う際に、レコード処理装置が過負荷になることを防ぎ、分散コンピューティング環境における動作を考慮してレコード処理装置が自らのタイミングで処理の制御（処理の実行、停止、再起動）ができる特性を保ちつつ、レコード処理装置の数が限られた中で少なくとも１つ以上のレコード処理装置が他のレコード処理装置と何らかの情報共有または情報交換をすることなく、独立かつ並行にレコード蓄積装置からレコードを受信して処理を行いながら、レコード蓄積装置からの配信（Ｐｕｓｈ）でレコードを受信するレコード処理装置と自らレコード蓄積装置からレコードを取得（Ｐｕｌｌ）するレコード処理装置の割合を当該時点においてＰｕｓｈモードでレコード受信しているレコード処理装置の割合と当該時点における第１のキュー使用率に応じてシステム無停止で動的に変更することで、キューに蓄積された各レコードの処理完了までにかかる時間を短くする（例えば、図９（Ｄ）のような）高効率な分散データ処理システム及び方法及びプログラムに関する。
【課題を解決するための手段】
【０００８】
図１は、本発明の原理構成図である。
【０００９】
本発明（請求項１）は、少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードを第１のキューに蓄積するレコード蓄積装置と、該レコード蓄積装置からレコードを受信して処理を行うレコード処理装置とが分散配置されており、
少なくとも１つ以上のレコード処理装置が、他のレコード処理装置と何らかの情報共有または情報交換をすることなく、独立かつ並行にレコード蓄積装置からレコードを受信して処理を行う分散データ処理システムであって、
レコード蓄積装置１００は、
第１のキュー１２０と、
クライアントから受信したレコードにレコード識別情報を付加してレコードの到着順（ＦＩＦＯ）に第１のキュー１２０に蓄積する第１のレコード蓄積手段１１０と、
第１のキュー１２０から蓄積したレコードを、レコードが到着した順（ＦＩＦＯ）に取り出す第１のレコード抽出手段１３０と、
取り出したレコードをレコード処理装置に送信するレコード配信手段１４０と、
を具備し、
レコード処理装置２００は、
第２のキュー２３０と、
レコード蓄積装置１００からレコードを取得するレコード取得手段２１０と、
取得したレコードを第２のキュー２３０に到着順（ＦＩＦＯ）に蓄積する第２のレコード蓄積手段２２０と、
第２のキュー２３０から蓄積したレコードをレコードの到着順（ＦＩＦＯ）に取り出す第２のレコード抽出手段２４０と、
取得したレコードのレコード識別情報に応じて、予め決められた処理を実行するレコード処理手段２５０と、
第２のレコード抽出手段２４０によって取り出したレコードをレコード処理手段２５０によって処理し、レコード処理完了後に、第２のレコード抽出手段２４０によって第２のキュー２３０からレコードを取り出す処理を繰り返す手段を含む。
【００１０】
また、本発明（請求項２）は、レコード処理装置２００において、
レコード蓄積装置１００からレコードを受信する前に、該レコード蓄積装置１００のレコード配信手段１４０から配信されたレコードを取得するＰｕｓｈモード、または、レコード取得手段により自律的にレコードを取得するＰｕｌｌモードのいずれか一方により、レコードを受信することを該レコード蓄積装置に通知するモード通知手段を有し、
レコード蓄積装置１００は、
レコード配信手段１４０によりレコードを受信するレコード処理装置の識別子を記録するＰｕｓｈモードメンバリスト記憶手段１５０と、
レコード取得手段２１０によりレコードを受信するレコード処理装置の識別子を記録するＰｕｌｌモードメンバリスト記憶手段１６０と、
レコード処理装置２００からのモードの通知に応じて、Ｐｕｓｈモードメンバリスト記憶手段１５０の登録状況と、Ｐｕｌｌモードメンバリスト記憶手段１６０の登録状況と、を更新するレコード受信モード制御手段１７０と、
少なくとも１つ以上のレコード処理装置２００が、レコード配信手段１４０またはレコード取得手段２１０のいずれか一方よりレコード蓄積装置からレコードを受信する際に、
Ｐｕｓｈモードの場合には、Ｐｕｓｈモードメンバリスト記憶手段１５０に含まれるレコード処理装置の中から何らかの方法で１つのレコード処理装置を選択し、選択したレコード処理装置の第２のキュー２３０に空きがあるかどうかを確認し、空きがあれば第１のレコード抽出手段によって第１のキュー１２０から取り出したレコードを配信し、Ｐｕｓｈモードメンバリスト記憶手段１５０に含まれる次のレコード処理装置に対して、同様の手順でレコードを配信する処理を繰り返す手段と、
Ｐｕｌｌモードの場合には、レコード処理手段２５０によるレコード処理完了後（当該レコード処理装置の第２のキュー２３０に処理待ちレコードが存在しなくなった時）に、当該レコード処理装置からレコード取得要求を受信し、該レコード処理装置がＰｕｌｌモードメンバリスト記憶手段１６０に含まれている場合に限り、第１のレコード抽出手段１３０によって第１のキューから取り出されたレコードを送信する処理を繰り返す手段と、
を含む。
【００１１】
また、本発明（請求項３）は、レコード処理装置２００において、
Ｐｕｓｈモードでレコードを受信しているとき、Ｐｕｌｌモードへ変更する第１のモード変更通知を、レコード蓄積装置に通知する第１のモード変更通知手段と、
Ｐｕｌｌモードでレコードを受信しているとき、レコード処理手段２５０によるレコード処理完了後に、Ｐｕｓｈモードへ変更する第２のモード変更通知を該レコード蓄積装置に通知する第２のモード変更通知手段と、を有し、
レコード蓄積装置１００のレコード受信モード制御手段１７０は、
第１のモード変更通知を受信すると、Ｐｕｓｈモードメンバリスト記憶手段１５０から通知元のレコード処理装置のエントリを削除し、Ｐｕｌｌモードメンバリスト記憶手段１６０の末尾に当該レコード処理装置のエントリを追加する手段と、
第２のモード変更通知を受信すると、Ｐｕｌｌモードメンバリスト記憶手段１６０から通知元のレコード処理装置のエントリを削除し、Ｐｕｓｈモードメンバリスト記憶手段１５０の末尾に当該レコード処理装置のエントリを追加する手段と、を含む。
【００１２】
また、本発明（請求項４）は、レコード受信モード制御手段１７０において、
第１のモード変更通知または第２のモード変更通知を取得すると、Ｐｕｓｈモードメンバリスト記憶手段１５０に登録されているレコード処理装置の数と、Ｐｕｌｌモードメンバリスト記憶手段１６０に登録されているレコード処理装置の数と、第１のキューに蓄積されている処理待ちレコードの数と、該第１のキュー１２０に蓄積可能な最大レコードの数からレコード受信モードの変更可否を決定する手段と、
レコード配信手段（Ｐｕｓｈモード）によってレコードを受信するレコード処理装置と、レコード取得手段（Ｐｕｌｌモード）によってレコードを受信するレコード処理装置の割合を動的に制御する手段と、を含む。
【００１３】
図２は、本発明の原理を説明するための図である。
【００１４】
本発明（請求項５）は、少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードを第１のキューに蓄積するレコード蓄積装置と、該レコード蓄積装置からレコードを受信して処理を行うレコード処理装置とが分散配置されており、少なくとも１つ以上のレコード処理装置が他のレコード処理装置と何らかの情報共有または情報交換をすることなく、独立かつ並行にレコード蓄積装置からレコードを受信して処理を行う分散データ処理システムにおける分散データ処理方法であって、
レコード蓄積装置は、
クライアントから受信したレコードにレコード識別情報を付加してレコードの到着順（ＦＩＦＯ）に第１のキューに蓄積し（ステップ１，２，３）、該第１のキューから蓄積したレコードを、レコードが到着した順（ＦＩＦＯ）に取り出し（ステップ４）、取り出したレコードをレコード処理装置に送信し（ステップ５）、
レコード処理装置は、
レコード蓄積装置からレコードを取得し、受信したレコードを到着順（ＦＩＦＯ）に第２のキューに蓄積し（ステップ６）、該第２のキューから蓄積したレコードをレコードの到着順（ＦＩＦＯ）に取り出し、レコードのレコード識別情報に応じて、予め決められた処理を実行し（ステップ７）、レコード処理完了後に、第２のキューからレコードを取り出す処理を繰り返す。
【００１５】
また、本発明（請求項６）は、レコード処理装置において、
レコード蓄積装置からレコードを受信する前に、該レコード蓄積装置のレコード配信手段からレコードを受信するＰｕｓｈモード、または、当該レコード処理装置のレコード取得手段により自律的に該レコード蓄積装置からレコードを受信するＰｕｌｌモードのいずれか一方により、レコードを受信することを該レコード蓄積装置に通知し、
レコード蓄積装置は、
Ｐｕｓｈモードである場合に、Ｐｕｓｈモードによりレコードを受信するレコード処理装置の識別子を記録するＰｕｓｈモードメンバリスト記憶手段に含まれるレコード処理装置の中から何らかの方法で１つのレコード処理装置を選択し、選択したレコード処理装置の第２のキューに空きがあるかどうかを確認し、空きがあれば第１のキューから取り出したレコードを配信し、Ｐｕｓｈモードメンバリスト記憶手段に含まれる次のレコード処理装置に対して、同様の手順でレコードを配信する処理を繰り返し、
Ｐｕｌｌモードである場合に、該レコード処理装置がレコード処理完了後（当該レコード処理装置の第２のキューに処理待ちレコードが存在しなくなった時）に、該レコード処理装置からレコードの取得要求があると、Ｐｕｌｌモードによりレコードを受信するレコード処理装置の識別子を記録するＰｕｌｌモードメンバリスト記憶手段に当該レコード処理装置が含まれている場合に限り、第１のキューからレコードを取り出し、取り出したレコードを該レコード処理装置に送信する処理を繰り返す。
【００１６】
また、本発明（請求項７）は、レコード処理装置が、Ｐｕｓｈモードでレコードを受信しているとき、Ｐｕｌｌモードへ変更する第１のモード変更通知を、レコード蓄積装置に通知し、
レコード蓄積装置が、第１のモード変更通知を受信すると、Ｐｕｓｈモードメンバリスト記憶手段から通知元のレコード処理装置のエントリを削除し、Ｐｕｌｌモードメンバリスト記憶手段の末尾に当該レコード処理装置のエントリを追加し、
また、レコード処理装置が、Ｐｕｌｌモードでレコードを受信しているとき、レコード処理手段によるレコード処理完了後に、Ｐｕｓｈモードへ変更する第２のモード変更通知を該レコード蓄積装置に通知し、
レコード蓄積装置が、第２のモード変更通知を受信すると、Ｐｕｌｌモードメンバリスト記憶手段から通知元のレコード処理装置のエントリを削除し、Ｐｕｓｈモードメンバリスト記憶手段の末尾に当該レコード処理装置のエントリを追加する。
【００１７】
また、本発明（請求項８）は、レコード蓄積装置において、
レコード処理装置からＰｕｓｈモードからＰｕｌｌモードへの変更の通知またはＰｕｌｌモードからＰｕｓｈモードへの変更の通知を取得すると、Ｐｕｓｈモードメンバリスト記憶手段に登録されているレコード処理装置の数と、Ｐｕｌｌモードメンバリスト記憶手段に登録されているレコード処理装置の数と、第１のキューに蓄積されている処理待ちレコードの数と、該第１のキューに蓄積可能な最大レコードの数からレコード受信モードの変更可否を決定する手段と、
Ｐｕｓｈモードによってレコードを受信するレコード処理装置と、Ｐｕｌｌモードによってレコードを受信するレコード処理装置の割合を動的に制御する。
【００１８】
本発明（請求項９）は、少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードを第１のキューに蓄積するレコード蓄積装置と、該レコード蓄積装置からレコードを受信して処理を行うレコード処理装置とが分散配置されており、少なくとも１つ以上のレコード処理装置が他のレコード処理装置と何らかの情報共有または情報交換をすることなく、独立かつ並行にレコード蓄積装置からレコードを受信して処理を行う分散データ処理システムにおけるレコード蓄積装置として機能するコンピュータに実行させるプログラムであって、
クライアントから受信したレコードにレコード識別情報を付加してレコードの到着順（ＦＩＦＯ）に第１のキューに蓄積し、該第１のキューから蓄積したレコードを、レコードが到着した順（ＦＩＦＯ）に取り出し、取り出したレコードをレコード処理装置に送信するレコード送信ステップをコンピュータに実行させ、
レコード送信ステップは、
レコード処理装置から、レコード蓄積装置からレコードが配信されるＰｕｓｈモードまたは、該レコード処理装置において該レコード蓄積装置よりレコードを取得するＰｕｌｌモードのいずれかのモードが指定されると、
Ｐｕｓｈモードの場合は、Ｐｕｓｈモードでレコードを受信するレコード処理装置が登録されているＰｕｓｈモードメンバリスト記憶手段を検索して、何らかの方法で１つのレコード処理装置を選択し、選択したレコード処理装置の第２のキューに空きがあるかどうかを確認し、空きがあれば第１のキューから取り出したレコードを配信し、Ｐｕｓｈモードメンバリスト記憶手段に含まれる次のレコード処理装置に対して、同様の手順でレコードを配信する処理を繰り返すステップと、
Ｐｕｌｌモードの場合は、Ｐｕｌｌモードでレコードを受信するレコード処理装置が登録されているＰｕｌｌモードメンバリスト記憶手段を検索して、当該レコード処理装置がＰｕｌｌモードメンバリストに含まれている場合に限り、第１のキューからレコードを取り出し、取り出したレコードを送信する処理を繰り返すステップと、を含むプログラムである。
【００１９】
本発明（請求項１０）は、少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードを第１のキューに蓄積するレコード蓄積装置と、該レコード蓄積装置からレコードを受信して処理を行うレコード処理装置とが分散配置されており、少なくとも１つ以上のレコード処理装置が他のレコード処理装置と何らかの情報共有または情報交換をすることなく、独立かつ並行にレコード蓄積装置からレコードを受信して処理を行う分散データ処理システムにおける該レコード処理装置として機能するコンピュータに実行させるプログラムであって、
レコード蓄積装置からレコードを受信する前に、該レコード蓄積装置からレコードが配信されるＰｕｓｈモードまたは、該レコード処理装置において該レコード蓄積装置よりレコードを取得するＰｕｌｌモードのいずれかのモードにより、レコードを受信することを該レコード蓄積装置に通知するモード通知ステップと、
レコード蓄積装置からレコードを取得して、取得した順（ＦＩＦＯ）に第２のキューに蓄積するレコード取得ステップと、
第２のキューから取得したレコードのレコード識別情報に応じて、予め決められた処理を実行するレコード処理ステップと、をコンピュータに実行させるプログラムである。
【発明の効果】
【００２０】
上述のように本発明によれば、クライアントから不規則な間隔で入力されるレコードを第１のキューに蓄積するレコード蓄積装置と、レコード蓄積装置から受信したレコードを第２のキューに蓄積して処理するレコード処理装置とが存在する分散コンピューティング環境上に、レコード蓄積装置と、レコード配信手段によりレコード蓄積装置からレコードを受信するレコード処理装置と、を分散配置し、レコード蓄積装置がクライアントから受信したレコードを第１のキューに蓄積し、レコード蓄積装置側のレコード配信手段によりレコード蓄積装置からレコードを受信するレコード処理装置と、自らのレコード取得手段によりレコード蓄積装置からレコードを受信するレコード処理装置とが存在する中で、レコード蓄積装置によるレコード配信手段と、レコード処理装置によるレコード処理手段とが同時並行に動作することで、系内の全てのレコード処理装置が過負荷になることを防ぎ、レコード取得手段によりレコードを受信しているレコード処理装置が自らのタイミングで処理の制御（処理の実行、停止、再起動）ができる特性を保ちつつ、レコード処理装置がレコードを処理する裏側で並行してレコード蓄積装置からレコードを受信できるようになり、レコード処理装置のレコード受信にかかる処理時間が隠蔽される。
【００２１】
その上、レコード処理装置がレコード配信手段によって送信されたレコードを受信し、第２のキューに蓄積してから処理することで、第２のキューがレコード処理の先読みバッファとなり、レコード処理装置がレコード蓄積装置からレコードを受信するまでの待ち時間がなくなり、連続してレコード処理を行えるようになり、レコード１件の取得にかかる時間が比較的長い場合やレコード１件の処理にかかる時間が比較的短い場合でも、キューに蓄積された各レコードの処理完了までにかかる時間を短くすることが可能である。
【００２２】
また、レコード配信手段による送信によりレコード蓄積装置からレコード受信しているレコード処理装置が、レコード取得手段によりレコードを受信することをレコード蓄積装置に通知し、通知を受けたレコード蓄積装置がＰｕｓｈモードメンバリストから当該レコード処理装置のエントリを削除し（これ以降はレコード配信手段でレコードが送信されてこなくなる）、Ｐｕｌｌモードメンバリストの末尾に当該レコード処理装置のエントリを追加し、当該レコード処理装置が、レコード処理手段によるレコード処理完了後（当該レコード処理装置の第２のキューに処理待ちレコードが存在しなくなった後）に、レコード取得手段によりレコードを受信することで、レコード蓄積装置からレコード配信手段によりレコードを受信しているレコード処理装置が、システムを停止することなく、レコード取得手段に変更してレコードを受信できる。
【００２３】
同様に、レコード取得手段によりレコード蓄積手段からレコードを受信しているレコード処理装置が、レコード処理手段によるレコード処理完了後（当該レコード処理装置の第２のキューに処理待ちレコードが存在しなくなった後）に、レコード配信手段によりレコードを受信することをレコード蓄積手段に通知し、通知を受けたレコード蓄積装置がＰｕｌｌモードメンバリストから当該レコード処理装置のエントリを削除し、Ｐｕｓｈモードメンバリストの末尾に当該レコード処理装置のエントリを追加し、当該レコード処理装置がレコード配信手段によりレコードを受信することで、レコード蓄積装置からレコード取得手段によりレコードを受信しているレコード処理装置が、システムを停止することなく、レコード配信手段に変更してレコードを受信できる。
【００２４】
更に、レコード処理装置からレコード受信モードの変更通知を受けたレコード蓄積装置がＰｕｓｈモードメンバリストにエントリされているレコード処理装置の数とＰｕｌｌモードメンバリストにエントリされているレコード処理装置の数と第１のキューに蓄積されている処理待ちレコードの数と第１のキューに蓄積可能な最大レコード数に応じてレコード受信モードの変更可否を決定することで、レコード処理装置の数が限られた中で少なくとも１つ以上のレコード処理装置が他のレコード処理装置と何らかの情報共有または情報交換をすることなく、レコード処理装置の自律性を保持したままで系内のＰｕｓｈ／Ｐｕｌｌ各レコード受信モードで動作するレコード処理装置の割合を動的に制御でき、上記効果を組み合わせることで、キューに蓄積された各レコードの処理完了までにかかる時間を短くする高効率な分散データ処理システムを実現することが可能である。
【発明を実施するための最良の形態】
【００２５】
以下、図面と共に本発明の実施の形態を説明する。
【００２６】
図３は、本発明の一実施の形態における分散データ処理システム構成を示す。
【００２７】
同図では、少なくとも１つ以上のクライアントから不規則な間隔で入力されるレコードを蓄積する第１のキュー１２０を持った少なくとも１つ以上のレコード蓄積装置１００と第２のキュー２３０を持った１つ以上のレコード処理装置２００からなる分散データ処理システムの構成を示す。なお、同図では、１つのレコード蓄積装置１００に対して複数のレコード処理装置２００が存在する構成例であるが、レコード蓄積装置１００が複数存在する構成も同様に実施可能である。
【００２８】
レコード蓄積装置１００は、第１のレコード蓄積部１１０、第１のキュー１２０、第１のレコード抽出部１３０、レコード配信部１４０、Ｐｕｓｈモードメンバリスト記憶部１５０、Ｐｕｌｌモードメンバリスト記憶部１６０、レコード受信モード制御部１７０から構成される。
【００２９】
レコード蓄積装置１００の第１のレコード蓄積部１１０は、クライアントから受信したレコードを蓄積し、レコード識別情報を付加してレコードの到着順（ＦＩＦＯ）に第１のキュー１２０に蓄積する。
【００３０】
第１のレコード抽出部１３０は、第１のキュー１２０からレコードを抽出し、レコード配信部１４０、または、レコード処理装置２００のレコード取得部２１０に渡す。
【００３１】
レコード配信部１４０は、Ｐｕｓｈモードでよって抽出されたレコードをレコード処理装置２００に配信する。
【００３２】
Ｐｕｓｈモードメンバリスト記憶部１５０は、Ｐｕｓｈモードメンバリストを管理する。Ｐｕｓｈモードメンバリストには、レコード配信部１４０によりレコードを受信するメンバ（レコード処理装置）が記録されている。
【００３３】
Ｐｕｌｌモードメンバリスト記憶部１６０は、Ｐｕｌｌモードメンバリストを管理する。おり、Ｐｕｌｌモードメンバリストには、レコード取得部２１０によりレコードを受信するメンバ（レコード処理装置）が記録されている。
【００３４】
レコード受信モード制御部１７０は、レコード処理装置２００からレコード取得要求があった場合に、Ｐｕｌｌモードメンバリスト記憶部１６０に当該レコード処理装置が登録されているかを判定する機能、Ｐｕｓｈモードメンバリスト記憶部１５０、Ｐｕｌｌモードメンバリスト記憶部１６０の内容を更新する機能等を有する。
【００３５】
レコード処理装置２００は、レコード取得部２１０、第２のレコード蓄積部２２０、第２のキュー２３０、第２のレコード抽出部２４０、レコード処理部２５０から構成される。
【００３６】
レコード取得部２１０は、Ｐｕｌｌモードによってレコード蓄積装置１００の第１のレコード抽出部１３０から取得する。
【００３７】
第２のレコード蓄積部２２０は、レコード蓄積装置１００のレコード配信部１４０から取得したレコードまたは、レコード取得部２１０から取得したレコードを、当該レコードの到着順（ＦＩＦＯ）に第２のキュー２３０に蓄積する。
【００３８】
第２のレコード抽出部２４０は、到着順（ＦＩＦＯ）に第２のキュー２３０からレコードを抽出し、レコード処理部２５０に渡す。
【００３９】
レコード処理部２５０は、取得したレコードのレコード識別情報に応じて予め決められた処理を実行する。
【００４０】
以下に上記の構成における動作を説明する。
【００４１】
図４は、本発明の一実施の形態におけるレコード蓄積装置の動作のフローチャートである。
【００４２】
同図では、レコード蓄積装置１００がクライアントから受信したレコードを第１のキュー１２０に蓄積する動作、及び、第１のキュー１２０に蓄積したレコードをレコード処理装置２００に提供する動作を示している。
【００４３】
以下では、レコード蓄積装置１００がレコードを蓄積する際の動作「動作１」、及び、レコード蓄積装置がレコード処理装置にレコードを提供する際の動作「動作２」「動作３」を示す。なお、レコード蓄積装置１００がレコードを第１のキュー１２０に蓄積する動作と、レコード蓄積装置１００が第１のキュー１２０に蓄積したレコードを配信部１４０（Ｐｕｓｈモード）によりレコード処理装置に提供する動作とレコード蓄積装置１００が第１のキューに蓄積したレコードをレコード取得部２１０（Ｐｕｌｌモード）によりレコード処理装置に提供する動作はそれぞれ独立並行に動作する。
【００４４】
同図における「動作１」では、レコード蓄積装置１００がクライアントから受信したレコードを第１のキュー１２０に蓄積する動作を示す。
【００４５】
ステップ１０１）  レコード蓄積装置１００の第１のレコード蓄積部１１０は、クライアントからレコードを受信するまで待機する。
【００４６】
ステップ１０２）  レコードを受信すると、第１のレコード蓄積部１１０は、当該レコードにレコード識別情報（例えば、クライアント識別情報、レコード取得タイムスタンプ、連番などから構成される情報）を付加する。
【００４７】
ステップ１０３）  第１のレコード蓄積部１１０は、第１のキュー１２０に空きがあるかをチェックし、空きがなければ空きができるまで待機する。
【００４８】
ステップ１０４）  第１のキュー１２０に空きがあれば、第１のキュー１２０にレコードの到着順（ＦＩＦＯ）にレコードを蓄積し、ステップ１０１からステップ１０４を繰り返す。
【００４９】
同図における「動作２」では、第１のキュー１２０に蓄積したレコードをレコード蓄積装置１００自らのレコード配信部１４０（Ｐｕｓｈモード）によりレコード処理装置に提供する動作を示す。
【００５０】
ステップ２０１）  第１のキュー１２０にレコードが存在しなければ、レコードが蓄積されるまで待機する。
【００５１】
ステップ２０２）  第１のキュー１２０にレコードが存在すれば、第１のレコード抽出部１３０が、第１のキュー１２０からレコードの到着順（ＦＩＦＯ）にレコードを取り出す。
【００５２】
ステップ２０３）  レコード配信部１４０は、Ｐｕｓｈモードメンバリスト記憶部１５０のＰｕｓｈモードメンバリストに含まれるレコード処理装置の中から、何らかの方法（例えば、ラウンドロビン順など）で１つのレコード処理装置２００を選択する。
【００５３】
ステップ２０４）  レコード配信部１４０は、選択したレコード処理装置２００に対して、当該レコード処理装置２００の第２のキュー２３０に空きがあるかを問い合わせ、当該レコード処理装置２００の第２のキュー２３０に空きがなければ、当該レコード処理装置２００へのレコード配信をスキップして、ステップ２０３に戻り、当該レコード処理装置２００の第２のキュー２３０に空きがあれば、ステップ２０５に移行する。
【００５４】
ステップ２０５）  レコード配信部１４０は、選択したレコードを処理装置２００にレコードを配信する。上記のステップ２０１〜２０５を繰り返す。
【００５５】
「動作３」は、第１のキュー１２０に蓄積したレコードをレコード処理装置２００のレコード取得部２１０（Ｐｕｌｌモード）によりレコード処理装置に提供する動作を示す。
【００５６】
ステップ３０１）  レコード蓄積装置１００は、レコード処理装置２００からレコード処理取得要求を受信するまで待機する。
【００５７】
ステップ３０２）  第１のレコード抽出部１３０は、レコード処理装置２００からレコード取得要求を受信すると、Ｐｕｌｌモードメンバリスト記憶部１６０を参照して、当該レコード処理装置２００がＰｕｌｌモードメンバリストに含まれているかを調べ、Ｐｕｌｌモードメンバリストに含まれていなければ、エラーを通知して、ステップ３０１に移行する。含まれていれば、次のステップに移行する。
【００５８】
ステップ３０３）  第１のレコード抽出部１３０は、第１のキュー１２０にレコードが存在するかをチェックし、レコードが存在しなければ、ステップ３０１に移行する。レコードが存在すれば、次のステップに移行する。
【００５９】
ステップ３０４）  第１のレコード抽出部１３０は、第１のキュー１２０からレコードの到着順（ＦＩＦＯ）にレコードを取り出す。
【００６０】
ステップ３０５）  取り出したレコードをレコード処理装置２００のレコード取得部２１０に提供する。ステップ３０１〜３０５を繰り返す。
【００６１】
次に、レコード処理装置２００がレコード蓄積装置１００からレコードを受信して、第２のキュー２３０に蓄積する動作及び第２のキュー２３０からレコードを取り出して、レコードを処理する動作について説明する。
【００６２】
図５は、本発明の一実施の形態におけるレコード処理装置の動作のフローチャートである。
【００６３】
同図において、レコード処理装置２００が自らのレコード取得部２１０によって受信したレコードを第２のキュー２３０に蓄積する際の動作（Ｐｕｌｌモード「動作４」）、及び、レコード蓄積装置１００側のレコード配信部１４０（Ｐｕｓｈモード）または、レコード処理装置２００の自らのレコード取得部２１０（Ｐｕｌｌモード）のいずれか一方によって受信したレコードを処理する際の動作（「動作５」）を示す。
【００６４】
なお、レコード蓄積装置１００側のレコード配信部１４０（Ｐｕｓｈモード）によって受信したレコードを第２のキュー２３０に蓄積する動作と、または、自らのレコード取得部２１０（Ｐｕｌｌモード）のいずれか一方によって受信したレコードを処理する動作はそれぞれ独立並行に動作する。
【００６５】
最初に、レコード処理装置２００の第２のレコード蓄積部２２０が、レコード蓄積装置１００のレコード配信部１４０から受信したレコードを第２のキュー２３０に蓄積する動作（「動作４」）について説明する。
【００６６】
ステップ４０１）  第２のレコード蓄積部２２０は、自身がＰｕｓｈモードで動作しているかＰｕｌｌモードで動作しているかをチェックし、Ｐｕｌｌモードで動作している間は何も行わない。Ｐｕｓｈモードで動作していれば、次のステップに移行する。
【００６７】
ステップ４０２）  レコード蓄積装置１００からレコードを受信していなければステップ４０１に移行する。レコードを受信したら次のステップに移行する。
【００６８】
ステップ４０３）  第２のレコード蓄積部２２０は、第２のキュー２３０に空きがあるかをチェックし、空きがなければ空きができるまで待機する。空きがあれば次のステップに移行する。
【００６９】
ステップ４０４）  第２のキュー２３０にレコードの到着順（ＦＩＦＯ）にレコードを蓄積する。ステップ４０１〜４０４を繰り返す。
【００７０】
次に、第２のレコード蓄積部２２０、または、レコード処理装置２００のレコード取得部２１０のいずれか一方によって受信したレコードを処理する動作（「動作５」）について説明する。
【００７１】
ステップ５０１）  レコード処理装置２００がＰｕｓｈモードで動作しているか、Ｐｕｌｌモードで動作しているかをチェックし、Ｐｕｓｈモードで動作している場合は、ステップ５０２に移行し、Ｐｕｌｌモードで動作している場合はステップ５０３に移行する。
【００７２】
ステップ５０２）  Ｐｕｓｈモードで動作している場合、第２のキュー２３０にレコードが存在するかをチェックし、レコードが存在しなければステップ５０１に戻る。レコードが存在していればステップ５０７に移行する。
【００７３】
ステップ５０３）  Ｐｕｌｌモードで動作している場合、第２のキュー２３０にレコードが存在するかチェックし、レコードが存在していれば、ステップ５０７に移行する。レコードが存在しなければ次のステップに移行する。
【００７４】
ステップ５０４）  レコード蓄積装置１００へレコード取得要求を送信する。
【００７５】
ステップ５０５）  第１のキュー１２０からレコードを受信できなければ、ステップ５０１に戻る。レコードを受信できれば、次のステップに移行する。
【００７６】
ステップ５０６）  第２のレコード蓄積部２２０は、第２のキュー２３０にレコードの到着順（ＦＩＦＯ）にレコードを蓄積する。
【００７７】
ステップ５０７）  第２のレコード抽出部２４０は、第２のキュー２３０からレコードの到着順（ＦＩＦＯ）にレコードを取り出す。
【００７８】
ステップ５０８）  レコード処理部２５０は、レコード識別情報に応じて予め決められた処理を実行する。ステップ５０１からステップ５０８の処理を繰り返す。
【００７９】
次に、モードの変更について説明する。
【００８０】
図６は、本発明の一実施の形態におけるモード変更（Ｐｕｓｈ→Ｐｕｌｌ）のシーケンスチャートである。
【００８１】
同図では、レコード蓄積装置１００からレコード蓄積装置１００側のレコード配信部１４０（Ｐｕｓｈモード）によりレコードを受信しているレコード処理装置２００が、レコード取得部２１０（Ｐｕｌｌモード）に変更してレコードを受信する動作を示している。
【００８２】
ステップ６０１）  レコード処理装置２００は、レコード蓄積装置１００側のレコード配信部１４０（Ｐｕｓｈモード）によりレコード受信することをレコード蓄積装置１００に通知する。
【００８３】
ステップ６０２）  通知を受けたレコード蓄積装置１００は、Ｐｕｌｌモードメンバリスト記憶部１６０のＰｕｌｌモードメンバリストから当該レコード処理装置のエントリを削除し、Ｐｕｓｈモードメンバリスト記憶部１５０のＰｕｓｈモードメンバリストの末尾に当該レコード処理装置のエントリを追加する。
【００８４】
ステップ６０３）  レコード蓄積装置１００がレコード処理装置２００にレコードを配信する。これ以降、当該レコード処理装置２００に対するレコード蓄積装置１００側のレコード配信部１４０（Ｐｕｓｈモード）によるレコード配信が開始される。
【００８５】
ステップ６０４）  レコード処理装置２００がレコードを受信する（レコード配信部１４０（Ｐｕｓｈモード））。
【００８６】
ステップ６０５）  レコード処理装置２００がレコード取得部２１０（Ｐｕｌｌモード）によりレコードを受信することをレコード蓄積装置１００に通知する。
【００８７】
ステップ６０６）  通知を受けたレコード蓄積装置１００がＰｕｓｈモードメンバリスト記憶部１５０のＰｕｓｈモードメンバリストから当該レコード処理装置のエントリを削除し、Ｐｕｌｌモードメンバリスト記憶部１６０のＰｕｌｌモードメンバリストの末尾に当該レコード処理装置２００のエントリを追加する。これ以降、当該レコード処理装置２００に対するレコード蓄積装置１００側のレコード配信部１４０（Ｐｕｓｈモード）によるレコードの配信が停止する。
【００８８】
ステップ６０７）  レコード処理装置２００は、第２のキュー２３０にレコードが存在していれば、レコードが存在しなくなるまで第２のキュー２３０からレコードを取り出し、レコード識別情報に応じて予め決められた処理を実行する。第２のキュー２３０にレコードが存在しなければレコード蓄積装置１００にレコード取得要求を送信する。
【００８９】
ステップ６０８）  レコード取得要求を受信したレコード蓄積装置１００がレコード処理装置２００にレコードを提供する。
【００９０】
ステップ６０９）  レコード処理装置２００のレコード取得部２１０（Ｐｕｌｌモード）がレコードを受信する。
【００９１】
次に、レコード蓄積装置１００からレコード処理装置２００のレコード取得部２１０（Ｐｕｌｌモード）によりレコードを受信しているレコード処理装置２００が、レコード蓄積装置１００側のレコード配信部１４０（Ｐｕｓｈモード）に変更してレコードを受信する動作を説明する。
【００９２】
図７は、本発明の一実施の形態におけるモード変更（Ｐｕｌｌモード→Ｐｕｓｈモード）のシーケンスチャートである。
【００９３】
ステップ７０１）  レコード処理装置２００がレコード取得部２１０（Ｐｕｌｌモード）によりレコードを受信することをレコード蓄積装置１００に通知する。
【００９４】
ステップ７０２）  通知を受けたレコード蓄積装置１００がＰｕｓｈモードメンバリスト記憶部１５０のＰｕｓｈモードメンバリストから当該レコード処理装置２００のエントリを削除し、Ｐｕｌｌモードメンバリスト記憶部１６０のＰｕｌｌモードメンバリストの末尾に当該レコード処理装置２００のエントリを追加する。
【００９５】
ステップ７０３）  レコード処理装置２００がレコード蓄積装置１００にレコード取得要求を送信する。
【００９６】
ステップ７０４）  レコード取得要求を受信したレコード蓄積装置１００がレコード処理装置２００にレコードを提供する。
【００９７】
ステップ７０５）  レコード処理装置２００がレコード取得部２１０（Ｐｕｌｌモード）でレコードを受信する。
【００９８】
ステップ７０６）  レコード処理装置２００は第２のキュー２３０にレコードが存在していれば、レコードが存在しなくなるまで第２のキュー２３０からレコードを取り出し、レコード識別情報に応じて予め決められた処理を実行する。第２のキュー２３０にレコードが存在しなければレコード蓄積装置１００側のレコード配信部１４０（Ｐｕｓｈモード）によりレコードを受信することをレコード蓄積装置１００に通知する。
【００９９】
ステップ７０７）  通知を受けたレコード蓄積装置１００がＰｕｌｌモードメンバリスト記憶部１６０のＰｕｌｌモードメンバリストから当該レコード処理装置２００のエントリを削除し、Ｐｕｓｈモードメンバリスト記憶部１５０のＰｕｓｈモードメンバリストの末尾に当該レコード処理装置２００のエントリを追加する。
【０１００】
ステップ７０８）  レコード蓄積装置１００がレコード処理装置２００にレコードを配信する。これ以降、当該レコード処理装置２００に対するレコード配信部１４０（Ｐｕｓｈモード）によるレコード配信が開始される。
【０１０１】
次に、レコード蓄積装置１００がレコード受信モードの変更可否を決定する動作を説明する。
【０１０２】
図８は、本発明の一実施の形態におけるレコード蓄積装置がレコード受信モードの変更可否を決定する動作のフローチャートである。
【０１０３】
同図では、レコード処理装置２００からレコード受信モードの変更（Ｐｕｓｈモード／Ｐｕｌｌモードの変更）通知を受けたレコード蓄積装置１００が、Ｐｕｓｈモードメンバリスト記憶部１５０のＰｕｓｈモードメンバリストにエントリされているレコード処理装置２００とＰｕｌｌモードメンバリスト記憶部１６０のＰｕｌｌモードメンバリストにエントリされているレコード処理装置２００の数と第１のキュー１２０に蓄積されている処理待ちレコードの数と第１のキュー１２０に蓄積可能な最大レコード数からレコード受信モードの変更可否を決定する動作を示している。
【０１０４】
ステップ８０１）  レコード蓄積装置１００は、レコード処理装置２００からレコード受信モードの変更通知を受信する。
【０１０５】
ステップ８０２）  レコード受信モード制御部１７０は、Ｐｕｓｈモードメンバリスト記憶部１５０のＰｕｓｈモードメンバリストにエントリされているレコード処理装置２００の数ａを取得する。
【０１０６】
ステップ８０３）  Ｐｕｌｌモードメンバリスト記憶部１６０のＰｕｌｌモードメンバリストにエントリされているレコード処理装置２００の数ｂを取得する。
【０１０７】
ステップ８０４）  レコード受信モード制御部１７０は、第１のキュー１２０に蓄積されているレコードの数を取得する。
【０１０８】
ステップ８０５）  第１のキュー１２０に蓄積可能なレコード数を取得する。
【０１０９】
ステップ８０６）  ステップ８０２のａ及びステップ８０３のｂからＰｕｓｈモードでレコードを受信しているレコード処理装置２００の割合を計算する。
【０１１０】
ステップ８０７）  ステップ８０４及びステップ８０５から第１のキュー１２０の使用率（処理待ちレコー数／最大レコード数）を計算する。
【０１１１】
ステップ８０８）  上記のステップ８０６及びステップ８０７からレコード受信モードの変更条件を満たしているか判断する。レコード受信モードの変更を許可するための条件の例を以下に示す。
【０１１２】
１）  第１のキューの使用率が８０％未満かつ変更後にＰｕｓｈモードでレコード受信しているレコード処理装置の割合が全体の５０％以内であれば、Ｐｕｓｈ→Ｐｕｌｌへの変更／Ｐｕｌｌ→Ｐｕｓｈへの変更ともに許可する。
【０１１３】
２）  第１のキューの使用率が８０％以上であれば、Ｐｕｓｈ→Ｐｕｌｌへの変更のみ許可する。
【０１１４】
ステップ８０９）  レコード受信モード制御部１７０は、レコード受信モードの変更条件を満たしていれば、レコード受信モードの変更を許可し、レコード受信モードの変更条件を満たしていなければレコード受信モードの変更を拒絶する。
【０１１５】
図９は、本発明の一実施の形態におけるレコード処理装置の処理時間を示す。
【０１１６】
同図は、レコード処理装置２００がレコード蓄積装置１００からレコードを受信（抽出及び転送）するのにかかる時間（Ｐｕｓｈモードではレコード抽出及びレコード配信にかかる時間、Ｐｕｌｌモードではレコード抽出及びレコード取得にかかる時間）とレコードを処理するのにかかる時間が等しいと仮定した場合に８件のレコードを３つのレコード処理装置Ａ，Ｂ，Ｃが１つのレコード蓄積装置１００から受信（抽出及び転送）して処理完了するまでにかかる時間の例を示す。
【０１１７】
同図（Ａ）は、全てのレコード処理装置Ａ，Ｂ，ＣがＰｕｓｈモードでレコードを受信する場合を示しており、レコード蓄積装置１００からのレコード配信後にレコード処理装置のレコード処理を実行するため、合計で９単位時間かかっている。
【０１１８】
同図（Ｂ）は、３つのレコード処理装置Ａ，Ｂ，ＣがＰｕｌｌモードでレコード受信する場合を示しており、レコード取得とレコード処理が逐次的に行われるため、６単位時間掛かっている。
【０１１９】
同図（Ｃ）は、２つのレコード処理装置Ａ，ＢがＰｕｓｈモードでレコードを受信し、レコード処理装置ＣがＰｕｌｌモードでレコード受信する場合を示しており、６単位時間かかっている。
【０１２０】
同図（Ｄ）は、１つのレコード処理装置ＡがＰｕｓｈモードで受信し、２つのレコード処理装置Ｂ，ＣがＰｕｌｌモードでレコード受信する場合を示しており、５単位時間しかかかっていない。
【０１２１】
上記の（Ｃ）及び（Ｄ）のようにＰｕｓｈモードのレコード処理装置とＰｕｌｌモードのレコード処理装置が混在する場合でも、Pushモード／Ｐｕｌｌモードの割合に応じてレコード受信（抽出及び転送）から処理完了までにかかる時間が異なることを示している。
【０１２２】
この例では、Ｐｕｓｈモードのレコード処理装置２００とＰｕｌｌモードのレコード処理装置２００の割合を予め設定した固定の割合で動作させるのではなく、クライアントからの不規則な入力に対応するために当該時点において、Ｐｕｓｈモードでレコード受信しているレコード処理装置２００の割合と当該時点における第１のキュー１２０の使用率に応じて適切に設定することによってレコード処理装置がレコード蓄積装置１００からレコードを受信（抽出及び転送）して処理完了するまでにかかる時間が短縮される可能性を示している。
【０１２３】
なお、第１のキュー１２０、及び、第２のキュー２３０にそれぞれレコードカウンタを用意しておき、レコード蓄積時にレコードカウンタを１加算、レコード抽出時にレコードカウンタを１減算することで、各キュー１２０，２３０内のレコード数を管理することができ、各キュー１２０，２３０にレコードが存在するかをチェックすることができる。
【０１２４】
更に、各キュー１２０，２３０に蓄積可能な最大レコード数と各レコードカウンタの値を比較することで各キュー１２０，２３０に空きがあるかを調べることも可能である。
【０１２５】
なお、上記のレコード蓄積装置の構成要素の動作をプログラムとして構築し、レコード蓄積装置として利用されるコンピュータにインストールして実行させる、または、ネットワークを介して流通させることが可能である。
【０１２６】
同様に、レコード処理装置の構成要素の動作をプログラムとして構築し、レコード処理装置として利用されるコンピュータにインストールして実行させる、または、ネットワークを介して流通させることが可能である。
【０１２７】
また、構築されたプログラムをハードディスクや、フレキシブルディスク・ＣＤ−ＲＯＭ等の可搬記憶媒体に格納し、コンピュータにインストールする、または、配布することが可能である。
【０１２８】
なお、本発明は、上記の実施の形態に限定されることなく、特許請求の範囲内において種々変更・応用が可能である。
【産業上の利用可能性】
【０１２９】
本発明は、１つ以上のクライアントから不規則な間隔で入力されるレコードをキューに蓄積しているレコード蓄積主体と、当該レコード蓄積主体からレコードを取得して処理を行うレコード処理主体とが分散配置されているシステムに適用可能である。
【図面の簡単な説明】
【０１３０】
【図１】本発明の原理構成図である。
【図２】本発明の原理を説明するための図である。
【図３】本発明の一実施の形態における分散データ処理システムの構成図である。
【図４】本発明の一実施の形態におけるレコード蓄積装置の動作のフローチャートである。
【図５】本発明の一実施の形態におけるレコード処理装置の動作のフローチャートである。
【図６】本発明の一実施の形態におけるモード変更（Ｐｕｓｈ→Ｐｕｌｌ）のシーケンスチャートである。
【図７】本発明の一実施の形態におけるモード変更（Ｐｕｌｌ→Ｐｕｓｈ）のシーケンスチャートである。
【図８】本発明の一実施の形態におけるレコード蓄積装置がレコード受信モードの変更可否を決定する動作のフローチャートである。
【図９】本発明の一実施の形態におけるレコード処理装置の処理時間を示す図である。
【符号の説明】
【０１３１】
１００  レコード蓄積装置
１１０  第１のレコード蓄積手段、第１のレコード蓄積部
１２０  第１キュー
１３０  第１のレコード抽出手段、第１のレコード抽出部
１４０  レコード配信手段、レコード配信部
１５０  Ｐｕｓｈモードメンバリスト記憶手段、Ｐｕｓｈモードメンバリスト記憶部
１６０  Ｐｕｌｌモードメンバリスト記憶手段、Ｐｕｌｌモードメンバリスト記憶部
１７０  レコード受信モード制御手段、レコード受信モード制御部
２００  レコード処理装置
２１０  レコード取得手段、レコード取得部
２２０  第２のレコード蓄積手段、第２のレコード蓄積部
２３０  第２のキュー
２４０  第２のレコード抽出手段、第２のレコード抽出部
２５０  レコード処理手段、レコード処理部
【図１】
2010117942.tif 000003
【図２】
2010117942.tif 000004
【図３】
2010117942.tif 000005
【図４】
2010117942.tif 000006
【図５】
2010117942.tif 000007
【図６】
2010117942.tif 000008
【図７】
2010117942.tif 000009
【図８】
2010117942.tif 000010
【図９】
2010117942.tif 000011