Redis AS3クライアント

今回はActionScript3.0用のRedisクライアントについて調べてみました (参照: as3redis)。

結論から書くと、「使うのは控えた方が良い」ということになるかと思います。以下理由。
(太字になっている部分が特に重要)

古いプロトコルにしか対応していない
分散できない
パイプライニング対応部分の設計が良くない

上2つは機能が足りないという点で良くありません。対応プロトコルが古いというのは公開時期を考えると仕方ないことですが、今後新しいプロトコルに対応させないと動かなくなる可能性も考えられます。しかも古いプロトコルだとコマンドによってデータの組み立て方も変わってくるので内部でそれを覚えておかないといけません。さらに、Consistent Hashing などが実装されていないためキーを分散できません。イマドキのクライアントだとこれは辛いところです。

3つ目についてはパフォーマンスに関係するところなので改善は簡単かと思いますが一応。as3redisではコマンドをキュー(実装上ではVectorを使っている)で管理しており、一定時間内に複数のコマンドがキューにpushされた場合はパイプライニングで送信する仕様になっています。そのため、なんと時間を毎フレーム監視する実装になっています;; なのでここは他言語のクライアントの実装を見習い、パイプライニングでコマンドを送信するための別のAPIを提供しておいて、単一コマンドを送信するAPIはもっとシンプルで軽い実装にした方がいいと思います。

このライブラリはかなりFatな実装になっていて、1コマンド1クラスで構成されています。つまりコマンドを1つ送信するためにインスタンスを生成するのでメモリがたくさん必要になってしまいます。抽象度が高いと実装が読みやすくて助かるのですが、パフォーマンスを考えるともう少し泥臭く書いてもいいんじゃないかなと思いました。新しいプロトコルであれば、全てのコマンドでデータの組み立て方が統一されているのでそんなに汚くならないはずです。

ここではライブラリを使わずにRedisとデータをやりとりしてみます。

* キー mykey に対して INCR コマンドにより数値をインクリメントする (クライアント数 1000)

package {
  import flash.display.Sprite;
  import flash.errors.EOFError;
  import flash.errors.IOError;
  import flash.events.Event;
  import flash.events.IOErrorEvent;
  import flash.events.ProgressEvent;
  import flash.events.SecurityErrorEvent;
  import flash.net.Socket;
  import flash.utils.ByteArray;
  import flash.utils.IDataInput;
  import flash.utils.IDataOutput;
 
  public class RedisTest extends Sprite {
    private const HOST:String = 'example.com';
    private const PORT:int = 6379;
    private const NUM:int = 1000;
    private var pool:Vector.<Socket>;
    private var buffer:ByteArray;
    private var cmd:String;
        
    public function RedisTest():void {
      if(stage) init();
      else addEventListener(Event.ADDED_TO_STAGE, init);
    }
    private function init(e:Event = null):void {
      removeEventListener(Event.ADDED_TO_STAGE, init);
      pool = new Vector.<Socket>(NUM, true);
      buffer = new ByteArray();
      for(var i:int=0; i<NUM; i++) {
        pool[i] = new Socket(HOST, PORT);
        pool[i].addEventListener(Event.CONNECT, onConnect);
        pool[i].addEventListener(Event.CLOSE, onClose);
        pool[i].addEventListener(ProgressEvent.SOCKET_DATA, onProgress);
        pool[i].addEventListener(IOErrorEvent.IO_ERROR, onIOError);
        pool[i].addEventListener(SecurityErrorEvent.SECURITY_ERROR, onSecurityError);
      }   
    }
    private function onConnect(e:Event):void {
      var stream:IDataOutput = e.target as IDataOutput;  // as Socket
      cmd = '*2\r\n$4\r\nINCR\r\n$5\r\nmykey\r\n';
      stream.writeUTFBytes(cmd);
      (stream as Socket).flush();
    }
    private function onClose(e:Event):void {
      trace(e);
    }
    private function onProgress(e:ProgressEvent):void {
    // => INCRコマンドのリプライは ":数値文字列"  e.g) ":100"
      try {
        var stream:IDataInput = e.target as IDataInput;
        stream.readBytes(buffer, buffer.length, stream.bytesAvailable);
        trace(buffer.toString());
      }catch(e:EOFError) {
        trace(e);
      }catch(e:IOError) {
        trace(e);
      }
    }
    private function onIOError(e:IOErrorEvent):void {
      trace(e);
    }
    private function onSecurityError(e:SecurityErrorEvent):void {
      trace(e);
    }
  }
}

package {

import flash.display.Sprite;

import flash.errors.EOFError;

import flash.errors.IOError;

import flash.events.Event;

import flash.events.IOErrorEvent;

import flash.events.ProgressEvent;

import flash.events.SecurityErrorEvent;

import flash.net.Socket;

import flash.utils.ByteArray;

import flash.utils.IDataInput;

import flash.utils.IDataOutput;

public class RedisTest extends Sprite {

private const HOST:String = 'example.com';

private const PORT:int = 6379;

private const NUM:int = 1000;

private var pool:Vector.<Socket>;

private var buffer:ByteArray;

private var cmd:String;

public function RedisTest():void {

if(stage) init();

else addEventListener(Event.ADDED_TO_STAGE, init);

}

private function init(e:Event = null):void {

removeEventListener(Event.ADDED_TO_STAGE, init);

pool = new Vector.<Socket>(NUM, true);

buffer = new ByteArray();

for(var i:int=0; i<NUM; i++) {

pool[i] = new Socket(HOST, PORT);

pool[i].addEventListener(Event.CONNECT, onConnect);

pool[i].addEventListener(Event.CLOSE, onClose);

pool[i].addEventListener(ProgressEvent.SOCKET_DATA, onProgress);

pool[i].addEventListener(IOErrorEvent.IO_ERROR, onIOError);

pool[i].addEventListener(SecurityErrorEvent.SECURITY_ERROR, onSecurityError);

}

private function onConnect(e:Event):void {

var stream:IDataOutput = e.target as IDataOutput; // as Socket

cmd = '*2\r\n$4\r\nINCR\r\n$5\r\nmykey\r\n';

stream.writeUTFBytes(cmd);

(stream as Socket).flush();

}

private function onClose(e:Event):void {

trace(e);

}

private function onProgress(e:ProgressEvent):void {

// => INCRコマンドのリプライは ":数値文字列" e.g) ":100"

try {

var stream:IDataInput = e.target as IDataInput;

stream.readBytes(buffer, buffer.length, stream.bytesAvailable);

trace(buffer.toString());

}catch(e:EOFError) {

trace(e);

}catch(e:IOError) {

trace(e);

}

private function onIOError(e:IOErrorEvent):void {

trace(e);

}

private function onSecurityError(e:SecurityErrorEvent):void {

trace(e);

}

新しいリクエストプロトコル仕様は以下のようになっています。シンプルですね。
(プロトコル仕様 — redis v2.0.3　documentation)

*<引き数の数> CR LF
$<引き数1のバイト数> CR LF
<引き数データ> CR LF
...
$<引き数Nのバイト数> CR LF
<引き数データ> CR LF

*<引き数の数> CR LF

$<引き数1のバイト数> CR LF

<引き数データ> CR LF

...

$<引き数Nのバイト数> CR LF

<引き数データ> CR LF

INCR/DECR に限らずRedisのコマンドはアトミックに動作するので、複数プロセスから叩かれても整合性は保証されます。また、バイナリセーフなのでどんなデータでも保存できます。 ByteArrayのデータなどもそのまま送って大丈夫。

もし自分でクライアントライブラリを作る場合は、新しいプロトコルで話せるようにすることと、キーの分散に対応させることが必要かと思います。レスポンスを解釈するところは as3redis の実装を参考に。クライアントがFlashなので大切なデータを送らないようにして、あくまでキャッシュ用途で使うことになるかと思います。

flash.net.Socket を利用する際の注意点

サーバの管理者権限がないといろいろムリです。アフェリエイトとかで簡単に回収できるのでみんなVPS借りようぜ。
* ソケットポリシーファイルあれこれ (参照: ソケットポリシーファイルサーバの設定 | デベロッパーセンター)。

注意点としては、FlashPlayerのバージョンによってソケット接続のセキュリティポリシーが変わっているので調べておく必要があります (Flash Player 9および10におけるポリシーファイル関連の変更点 | デベロッパーセンター)。あと、ファイアウォールで843番ポートなんか閉じてると思うので、ソケットポリシーファイルを返すためにしぶしぶ開けておかなければなりません。iptables の場合は以下のように。

$ iptables -A INPUT -p tcp -m state --state NEW --dport 843 -j ACCEPT
$ /etc/init.d/iptables save
$ /etc/init.d/iptables restart

$ iptables -A INPUT -p tcp -m state --state NEW --dport 843 -j ACCEPT

$ /etc/init.d/iptables save

$ /etc/init.d/iptables restart

Flashの為にポート1個消費するとか納得いかない。。
そもそも”ポリシー”と呼んでいるものをころころ変えるっていうのはどうなの？

Redis AS3クライアント

flash.net.Socket を利用する際の注意点

あわせて読む:

コメントを残す