あおうさ＠日記

AppEngineでCounterを実装する #appengine #slim3

技術

※本エントリーのソースには誤りがありました。修正版はこちら

Counterの実装に興味がある人がいるようですのでアップしておきます。

仕様は
- increment()でカウントアップ
- カウントは日毎に持つ
- clear()でカウントを0に戻す

単純なカウンター

import java.io.Serializable;
import java.util.ConcurrentModificationException;
import java.util.Date;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.slim3.datastore.Attribute;
import org.slim3.datastore.Datastore;
import org.slim3.datastore.EntityNotFoundRuntimeException;
import org.slim3.datastore.Model;
import org.slim3.util.DateUtil;
import org.slim3.util.ThrowableUtil;

import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.Transaction;

@Model
public class Counter implements Serializable {

    private static final long serialVersionUID = 1L;

    @Attribute(primaryKey = true)
    private Key key;
    private Long seq;

    public Key getKey() {
        return key;
    }
    public void setKey(Key key) {
        this.key = key;
    }
    public Long getSeq() {
        return seq;
    }
    public void setSeq(Long seq) {
        this.seq = seq;
    }

    private static Logger logger = Logger.getLogger(Counter.class.getName());
    private static final int RETRY_MAX = 10;

    private static Key createKey(Date date) {
        String keyName = DateUtil.toString(date, DateUtil.ISO_DATE_PATTERN);
        return Datastore.createKey(Counter.class, keyName);
    }

    public static Long increment() {
        return increment(new Date());
    }

    public static Long increment(Date date) {
        Key key = createKey(date);
        Transaction tx = Datastore.beginTransaction();
        try {
            return createOrUpdate(tx, key);
        } catch (ConcurrentModificationException e) {
            for (int i = 0; i < RETRY_MAX; i++) {
                try {
                    return createOrUpdate(tx, key);
                } catch (ConcurrentModificationException e2) {
                    logger.log(Level.WARNING, "Retry("
                        + i
                        + "): "
                        + e2.getMessage(), e2);
                }
            }
            throw e;
        }
    }

    private static Long createOrUpdate(Transaction tx, Key key) {
        Counter counter = null;
        try {
            counter = Datastore.get(tx, Counter.class, key);
            counter.setSeq(counter.getSeq() + 1);
            Datastore.put(tx, counter);
        } catch (EntityNotFoundRuntimeException e) {
            counter = new Counter();
            counter.setKey(key);
            counter.setSeq(new Long(1));
            Datastore.put(tx, counter);
        }
        Datastore.commit(tx);
        return counter.getSeq();
    }

    public static void clear() {
        clear(new Date());
    }

    public static void clear(Date date) {
        Key key = createKey(date);
        Counter counter = Datastore.get(Counter.class, key);
        counter.setSeq(new Long(0));
        Datastore.put(counter);
    }

}

ShardingCounter

import java.io.Serializable;
import java.util.ConcurrentModificationException;
import java.util.Date;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.slim3.datastore.Attribute;
import org.slim3.datastore.Datastore;
import org.slim3.datastore.EntityNotFoundRuntimeException;
import org.slim3.datastore.Model;
import org.slim3.util.DateUtil;
import org.slim3.util.ThrowableUtil;

import com.blank.meta.ShardingCounterMeta;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.Transaction;

@Model
public class ShardingCounter implements Serializable {

    private static final long serialVersionUID = 1L;

    private static final int KIND_COUNT = 10;

    private static final ShardingCounterMeta meta = ShardingCounterMeta.get();

    @Attribute(primaryKey = true)
    private Key key;

    private Long seq;

    public Key getKey() {
        return key;
    }

    public void setKey(Key key) {
        this.key = key;
    }

    public Long getSeq() {
        return seq;
    }

    public void setSeq(Long seq) {
        this.seq = seq;
    }

    private static Logger logger =
        Logger.getLogger(ShardingCounter.class.getName());
    private static final int RETRY_MAX = 10;

    private static Key createKey(String kind, Date date) {
        String keyName = DateUtil.toString(date, DateUtil.ISO_DATE_PATTERN);
        return Datastore.createKey(kind, keyName);
    }

    private static Key createKey(Date date) {
        String kind =
            meta.getKind() + "$" + (((int) (Math.random() * (KIND_COUNT))) + 1);
        return createKey(kind, date);
    }

    public static Long increment() {
        return increment(new Date());
    }

    public static Long increment(Date date) {
        Key key = createKey(date);
        Transaction tx = Datastore.beginTransaction();
        try {
            createOrUpdate(tx, key);
        } catch (ConcurrentModificationException e) {
            for (int i = 0; i < RETRY_MAX; i++) {
                try {
                    createOrUpdate(tx, key);
                } catch (ConcurrentModificationException e2) {
                    logger.log(Level.WARNING, "Retry("
                        + i
                        + "): "
                        + e2.getMessage(), e2);
                }
            }
            throw e;
        }
        return getTotalCount(date);
    }

    private static void createOrUpdate(Transaction tx, Key key) {
        try {
            Entity entity = Datastore.get(tx, key);
            ShardingCounter counter = meta.entityToModel(entity);
            counter.setSeq(counter.getSeq() + 1);
            Datastore.put(tx, meta.modelToEntity(counter));
        } catch (EntityNotFoundRuntimeException e) {
            ShardingCounter counter = new ShardingCounter();
            counter.setKey(key);
            counter.setSeq(new Long(1));
            Datastore.put(tx, meta.modelToEntity(counter));
        }
        Datastore.commit(tx);
    }

    private static Long getTotalCount(Date date) {
        Long result = new Long(0);
        for (int i = 0; i < KIND_COUNT; i++) {
            Key key = createKey(meta.getKind() + "$" + (i + 1), date);
            ShardingCounter counter = null;
            try {
                Entity e = Datastore.get(key);
                counter = meta.entityToModel(e);
            } catch (EntityNotFoundRuntimeException ignore) {
                continue;
            }
            result += counter.getSeq();
        }
        return result;
    }

    public static void clear() {
        clear(new Date());
    }

    public static void clear(Date date) {
        for (int i = 0; i < KIND_COUNT; i++) {
            Key key = createKey(meta.getKind() + "$" + (i + 1), date);
            try {
                ShardingCounter counter =
                    meta.entityToModel(Datastore.get(key));
                counter.setSeq(new Long(0));
                Datastore.put(meta.modelToEntity(counter));
            } catch (EntityNotFoundRuntimeException ignore) {
                continue;
            }
        }
    }

}

Counter.classはgetした値は必ずユニークですが、ShardingCounterはgetした値はユニークにならない可能性があります。

ShardingCounterが遅い

そもそも同時アクセスが少ないアプリを今作成していたのでShardingCounterはいらなかったのですが、どうせなら作ってしまおうと実装してみた。ところが、負荷テストをしてみたところShardingCounterが結構遅いんですよね。まだ厳密に調査できていなのですが、おそらくgetTotalCountが遅いんじゃないかなと思われます。
　
追記：getTotalCountをparallel getにすると早くなりました。

ConcurrentModificationExceptionになったTransactionオブジェクトはrollbackが必要か？また、そのまま再利用はできるのか？ #appengine

技術

ConcurrentModificationExceptionになったTransactionオブジェクトはrollbackは不要
再利用は不可、新規にbeginTransactionが必要。

という訳で上記のエントリーのソースにはトランザクション処理に誤りがありました。

    Key key;
    @Before
    public void setup() throws Exception {
        Counter e = new Counter();
        e.setSeq(new Long(1));
        key = Datastore.put(e);
    }

    @Test
    public void トランザクションのテスト() throws Exception {
        Transaction tx1 = Datastore.beginTransaction();
        Transaction tx2 = Datastore.beginTransaction();
        Counter e1 = Datastore.get(tx1, Counter.class, key);
        Counter e2 = Datastore.get(tx2, Counter.class, key);
        assertThat(e1, is(notNullValue()));
        assertThat(e2, is(notNullValue()));

        e1.setSeq(e1.getSeq() + 1);
        Datastore.put(tx1, e1);
        Datastore.commit(tx1);

        try {
            e2.setSeq(e2.getSeq() + 1);
            Datastore.put(tx2, e2);
            Datastore.commit(tx2);
        } catch (ConcurrentModificationException ignore) {
            assertThat(tx2.isActive(), is(false));
            Transaction tx3 = Datastore.beginTransaction();
            e2 = Datastore.get(tx3, Counter.class, key);
            e2.setSeq(e2.getSeq() + 1);
            Datastore.put(tx3, e2);
            Datastore.commit(tx3);
        }
        assertThat(e2.getSeq(), is(new Long(3)));
    }