AtomikosTransactionsEssentials を PostgreSQL8.3で使う。

実装

APサーバをTransactionManagerの為だけに使うのは、正直嘘っぽいですなぁ。
しかし、JTA単独で、実装されているプロダクトってあんまみない様なｷｶﾞｽ。
僕が知らないだけかもしらんけども。
僕が知っている幾つかのは、まぁ、ちょっと、アレでナニ、と言うか文句が色々あったりなかったり。
と言うワケで、この週末は、AtomikosTransactionsを、ちょっと動かしてみた。
負荷テストとか、リカバリ周りの動作検証とか、そういう本当に重要な事はやってないので、注意。


OSS版と商用版とあるのだけど、OSS版は、Apache License Version2.0なので、結構気軽に使えるます。
商用版は、何か比較がサイトには、書いてあるのだけど、
僕には、商用版の機能が必要になる現実的な状況がイメージ出来ない感じ。
ま、それはいっか。


まずは、ダウンロード。

ここの、Dowonloadリンクを踏むと、謎の入力フォームが現れるので、
そいつにせっせと入力する。
入力が終ると、入力しておいたメールアドレスに、
ファイルをダウンロードする為のURLが書かれたメールがすっとんでくるます。
で、ダウンロード。


僕がダウンロードしたのは、

  • AtomikosTransactionsEssentials-3.3.4.zip

と言うファイル。

アーカイブを展開して、必要なファイルだけ抜く。

色々がっさり入っているのだけど、結局は

  • /AtomikosTransactionsEssentials-3.3.4/dist/transactions-essentials-all.jar
  • /AtomikosTransactionsEssentials-3.3.4/lib/jta.jar
  • /AtomikosTransactionsEssentials-3.3.4/examples/lib/commons-logging-1.0.4.jar
  • /AtomikosTransactionsEssentials-3.3.4/included/slf4j/slf4j-jcl-1.4.3.jar
  • /AtomikosTransactionsEssentials-3.3.4/lib/slf4j-api-1.4.3.jar

この辺だけ抜けば、動く環境が作れるます。
後の三つは、ロギングライブラリなので、既にお好みの構成があれば、それを使う感じで。


これに、PostgreSQL8.3のJDBCドライバのjarを加える。

こいつの中には、XADataSourceの実装が入っているので、そいつを使うます。
Atomikosの中には、XADataSource実装は入っていないので、
RDBベンダが配布しているXADataSource実装を使うのが正しい姿の様ですな。
特に2フェーズコミットとか、必要ないんだぜ？って方は、
普通のJDBCドライバだけあれば、それでおｋ。*1

次に初期設定。

AtomikosTransactionsEssentials-3.3.4.zipを解凍すると、

  • jta.properties

なんてファイルが入っているもす。
こいつを、これから動かすコードのクラスパスルートに配置しるます。
ま、とりあえずコピーすれば、何となくは動きます。
そこそこ、真っ当なデフォルト値が、大体設定される様にはなっているみたい。

さて、実験だ。

中のコードやら、サンプルコードやら見ていると、色々泣きそうになるのだけど、
要は、AtomikosDataSourceBeanと言う奴を使えば、基本的な設定を行う事が出来る。
こんな感じ。

private static AtomikosDataSourceBean setUpDataSource() throws SQLException {
  AtomikosDataSourceBean ads = new AtomikosDataSourceBean();

  // REQUIRED
  ads.setXaDataSourceClassName("org.postgresql.xa.PGXADataSource");
  Properties props = new Properties();
  props.setProperty("ServerName", "localhost");
  props.setProperty("PortNumber", "5432");
  props.setProperty("DatabaseName", "atomikos");
  props.setProperty("User", "postgres");
  props.setProperty("Password", "postgres");
  ads.setXaProperties(props);
  ads.setUniqueResourceName("jdbc/postgreSQL/xaDatasource");

  // OPTIONAL
  ads.setPoolSize(3);
  ads.setBorrowConnectionTimeout(60);
  return ads;
}

後で、出てくるけどUserTransationやUserTransactionManagerやらは、
こいつと内部的なSingletonインスタンスの共有によって、必要な情報をやり取りしているます。
同じ名前のクラス名が、違うパッケージにあったりして、確信が得られるまで、ほんに苦労したなり。
それについて、説明する気は無いのだけども。


REQUIREDから始まっている数行は、動作させる為の必須項目。
主にJDBCドライバ周りのそれっぽい設定やね。
UniqueResourceNameだけが、何と言うかJTAっぽいっつうか、XAResourceっぽいっつうか…ちょと不思議な感じだけども。
まぁ、何かそれっぽいのを付ければ良いみたい。


OPTIONALから始まる数行は、AtomikosDataSourceBeanが、
ConnectionPoolの設定を内部的に行う為のプロパティだに。
今回は面倒だから、余り細かく設定してないけど、そこそこ細かい設定が出来る様にはなっている。
ま、AtomikosDataSourceBeanは、便利なセットアップ用ユーティリティなんで、
ここで、出来ないようなもっと細かいチューニングが必要なら、中のコードを抜いて
必要な部分だけ直せば良いデス。


ちなみに、PostgreSQLはlocalhostで動いていて、atomikosってデータベースが作ってあるます。


後はもう、実験用のコードを沢山。
一応、いくつかポイントはあって、

  • javax.transaction.UserTransactionを使いたい時は、com.atomikos.icatch.jta.UserTransactionImpをnewする。
    処理は、内部で参照を抱えてるTransactionManagerへのファサードでしかないので、
    インスタンスを幾つ作っても、それっぽく動きます。
    極端な話、beginとcommitを別なインスタンスに対して呼び出しても、正しく動作します。
    逆に、UserTransactionImpのインスタンスを複数のスレッドで共有しても、ちゃんと動きます。
    Spring用のサンプルではそうなっとりました。
  • javax.transaction.TransactionManagerを使いたい時は、com.atomikos.icatch.jta.UserTransactionManagerをnewする。クラスの構造としては、大体UserTransactionImpと同じで、実体の皆様へ処理を委譲していえるだけです。
    こいつはまぁ、TransactionManagerなんで、Singletonぽく扱うのがヨロシイかとオモイマス。
    closeメソッドがあって、中で参照を持ってる実体の皆様を明示的に後始末できるトコロがポイントかな。
private static void processTx(DataSource ds) throws SystemException {
  UserTransaction tx = null;
  try {
    tx = new UserTransactionImp();
    Connection con = null;
    try {
      tx.setTransactionTimeout(60);
      tx.begin();

      con = insertAndSelect(ds);
    } finally {
      close(con);
    }
    tx.rollback();
  } catch (Exception e) {
    e.printStackTrace();
    tx.rollback();
  }
}

private static void processTxManager(DataSource ds) throws Exception {
  UserTransactionManager utm = new UserTransactionManager();
  try {
    utm.init();
    Connection con = null;
    try {
      utm.begin();
      con = insertAndSelect(ds);
    } finally {
      close(con);
    }
    utm.rollback();
  } finally {
    utm.close();
  }
}

private static void processTxNew(DataSource ds) throws Exception {
  UserTransactionManager utm = new UserTransactionManager();
  try {
    utm.init();
    Connection con = null;
    try {
      utm.begin();
      con = insertAndSelect(ds);
      Transaction tx = utm.suspend();

      Connection con2 = null;
      try {
        utm.begin();
        con2 = insertAndSelect(ds);
      } finally {
        close(con2);
      }
      utm.rollback();
      utm.resume(tx);
    } finally {
      close(con);
    }
    utm.rollback();
  } finally {
    utm.close();
  }
}

private static void processNestedTx(DataSource ds) throws Exception {
  UserTransaction tx = null;
  try {
    tx = new UserTransactionImp();
    Connection con = null;
    try {
      tx.setTransactionTimeout(60);
      tx.begin();
      con = insertAndSelect(ds);

      UserTransaction nested = null;
      try {
        nested = new UserTransactionImp();
        System.out.println("*** nested begin ****");
        nested.begin();
        Connection con2 = null;
        try {
          nested.setTransactionTimeout(30);
          nested.begin();
          con2 = insertAndSelect(ds);
        } finally {
          close(con2);
        }
      } catch (Exception e) {
        e.printStackTrace();
      }
      nested.rollback();
    } finally {
      close(con);
    }
    tx.rollback();
  } catch (Exception e) {
    e.printStackTrace();
    tx.rollback();
  }
}

private static void close(Connection con2) throws SQLException {
  if (con2 != null) {
    con2.close();
  }
}

private static Connection insertAndSelect(DataSource ds)
    throws SQLException {
  Connection con = ds.getConnection();
  PreparedStatement ps = con
      .prepareStatement("INSERT INTO \"EMP\" (name,timestamp)VALUES('hoge',now());");
  ps.execute();

  ps = con.prepareStatement("SELECT * FROM \"EMP\"");
  ResultSet rs = ps.executeQuery();
  while (rs.next()) {
    int i = 1;
    System.out.printf("id :[%d] name : [%s] timestamp [%s]\r\n", rs
        .getInt(i++), rs.getString(i++), rs.getTimestamp(i++));
  }
  return con;
}

public static void main(String[] args) throws Exception {
  AtomikosDataSourceBean ds = setUpDataSource();
  try {
    processTx(ds);
    System.out.println("======================");
    processTxManager(ds);
    System.out.println("======================");
    processTxNew(ds);
    System.out.println("======================");
    processNestedTx(ds);
    System.out.println("======================");
  } finally {
    ds.close();
  }
}

*1:いや、まぁ、僕もあんまり2フェーズコミットとかイラナイんだけどね…