分散環境でのPHPによるタスク管理 - Zend_Queue -

はじめに

　本連載では、PHP上で動作するアプリケーションフレームワークであるZend Frameworkについて紹介していきます。今回は分散アプリケーション間でのデータのやりとりを行うための方法の一つである「メッセージキュー」を利用するためのZend_Queueコンポーネントを紹介します。

　ユーザからデータを受け取り、それの処理をするようなサービスがあったとします。この処理に時間がかかるようであれば、例えばフロントエンドでは「データを受け付けました」とだけ表示し、バックエンドでデータを処理する仕組みが必要です。このフロントエンドとバックエンドのプロセスの間での通信を管理するのがメッセージキューで、フロントエンドから受け取った処理の依頼（メッセージ）を受付順に並べておき、要求に応じて先頭から順にバックエンドに送信します。

　Zend_Queueコンポーネントは、このメッセージキューを作成する機能を提供します。今回は、まず基本的なZend_Queueコンポーネントの使い方について見た後、キューの格納にMySQLとApache ActiveMQを利用する方法について見ていきたいと思います。

対象読者

　PHPの基本構文については一通り理解している方を対象としています。

必要な環境

　Zend FrameworkはPHP 5.2.4以降とWebサーバがインストールされている環境で利用可能です。本稿ではWebサーバとしてApache 2.2を、OSにWindows XPを採用し、アプリケーションを作成していきます。

　以下に、今回アプリケーション作成／動作確認に用いた環境を示します（インストールにあたっては最新安定版の使用を推奨します）。各項目の詳細なインストール手順は、「サーバサイド技術の学び舎 - WINGS」より｢サーバサイド環境構築設定手順｣を参照ください。

• Windows XP SP3
• PHP 5.3.3
• Apache 2.2.16

　LinuxやFreeBSDなどUNIX系OSをお使いの方もコマンドはほぼ一緒ですので、パスなどは適宜読み替えてください。

メッセージキューとZend_Queueの基本

　Zend_Queueはプロセスやスレッドの間でデータの通信をするための、メッセージキューを作成するコンポーネントです。メッセージキューを利用したデータの通信では、データ（メッセージ）を送る「送信者」、受け取る「受信者」と、メッセージを管理する「メッセージキュー」が存在します（図1）。

図1　メッセージキュー

　メッセージキュー内ではメッセージは基本的には登録された順番に並んでいて、最初に追加されたものから順に取得されるようになっています。そのため、すぐに対応する必要はないが、順番に対処して欲しい処理を管理するような場面に向いています。

　なお、図1を見て分かるとおり、送信者と受信者は直接通信せず、それぞれメッセージキューのみ通信をします。通常の同期型プロセス間通信では、データを送信するプロセスとデータを受信するプロセスの双方が同じタイミングで通信する必要があるため、通信待ち時間などの無駄が発生します。一方、メッセージキューを利用した場合は送信者と受信者が同期を取る必要がないため（非同期通信）、無駄な処理を減らすことができます。

配列を利用した簡単なサンプル

　それでは、まず一つのプロセス内でメッセージキューを使う例から見ていきましょう。メッセージをどのように格納するかにはいくつかの方法が選べるのですが、ここでは配列（Array）にメッセージを格納する方法を利用することにします。配列を利用する方法はプロセス間通信には使えませんが、簡単に利用できるためテストなどの用途には向いています（なお、格納方式の一覧については次のアダプタの説明を参照して下さい）。

　メッセージキューへの操作は大きく（1）メッセージキューの作成、（2）メッセージキューへのメッセージの送信、（3）メッセージキューからメッセージの受信の3種類に分けることができます（リスト1-3）。

<?php
require 'Zend/Queue.php';

/* メッセージキューの作成 */
/* オプションの指定 */
$options = array('name' => 'codezine_queue');//（1）
/* Arrayアダプタを利用したキューの作成 */
$queue = new Zend_Queue('Array', $options);//（2）

　ますメッセージキューを作成しています。（1）でメッセージキューへのオプションとして名前「codezine_queue」を指定した上で、（2）で配列を利用したメッセージキューを作成しています。

/* （3）メッセージの送信 */
$queue->send('job 1');
$queue->send('job 2');

　（3）では（1）で作成したメッセージキューにZend_Queueクラスのsendメソッドでメッセージを送信しています。ここではメッセージキューに「job 1」と「job 2」の 2つのメッセージが登録されました。

/* メッセージを最大 5つまで受信 */
$messages = $queue->receive(5);//（4）

/* それぞれのメッセージについて */
foreach ($messages as $i => $message) {
    /* （5）実際の処理 */
    echo $message->body, "\n";//ここではメッセージの中身の確認

    /* メッセージをキューから削除 */
    $queue->deleteMessage($message);//（6）
}

　▼

job 1
job 2

　（4）～（6）では（3）で送信されたメッセージを受信し、処理しています。まず（4）ではZend_Queueクラスのreceiveメソッドでメッセージを受信しています。ここで与えている引数の「5」は、最大5つのメッセージまで受信するように指示しています。この時点で受信されたメッセージは、受信済みとしてマークされ、次にreciveメソッドが呼ばれた時には、これに続くメッセージが受信されます。

　receiveメソッドの返り値はSPLイテレータを実装しているので、foreach文で中身の各メッセージにアクセスすることができます。（5）は各メッセージを実際に処理する部分です。ここではメッセージ本文にアクセスし出力していますが、実際にはより複雑な処理を行います。また（6）では処理済みのメッセージをメッセージキューから削除しています。

　なお、一回receiveメソッドで受信されたメッセージでも、一定時間deleteMessageメソッドで削除されない場合には受信されたことが取り消されます。この時間（タイムアウト時間）はキューごとにディフォルト値を指定することもできますし、receiveメソッドで渡すこともできます。receiveメソッドで指定する場合には2つ目の引数で指定します。例えばタイムアウト時間を10秒に指定する場合には「$queue->receive(5, 10)」とします。