@Test public void incr() throws Exception { String head = "incrA"; RKv kv = db.getrKv(); try { JoinFuture<String> joinFuture = JoinFuture.build(executorService, String.class); for (int i = 0; i < 10 * 10000; i++) { int fj = i; joinFuture.add(args -> { try { kv.incr(head, 1); } catch (Exception e) { e.printStackTrace(); } return ""; }); } joinFuture.join(); Assert.assertArrayEquals(ArrayKits.longToBytes(10 * 10000), kv.get(head)); } finally { kv.delPrefix(head); } }
public long incr(String key, int step, int ttl) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { byte[] keyb = getKey(key); LockEntity lockEntity = lock.lock(key); try { start(); byte[] key_b = ArrayKits.addAll(HEAD_B, keyb); byte[] value = get(key); long seq; if (value == null) { seq = step; } else { DAssert.isTrue(value.length == 8, ErrorType.DATA_LOCK, "value not a incr"); seq = ArrayKits.bytesToLong(value) + step; } putDB(key_b, ArrayKits.longToBytes(seq), SstColumnFamily.DEFAULT); int time = (int) (System.currentTimeMillis() / 1000 + ttl); putDB(ArrayKits.addAll(HEAD_TTL, keyb), ArrayKits.intToBytes(time), SstColumnFamily.DEFAULT); setTimer(KeyEnum.KV_TIMER, time, key_b); commit(); checkTxCommit(); return seq; } finally { lock.unlock(lockEntity); release(); } } catch (Exception e) { checkTxRollBack(); throw e; } }
RKv extends RBase { public long incr(String key, int step, int ttl) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { byte[] keyb = getKey(key); LockEntity lockEntity = lock.lock(key); try { start(); byte[] key_b = ArrayKits.addAll(HEAD_B, keyb); byte[] value = get(key); long seq; if (value == null) { seq = step; } else { DAssert.isTrue(value.length == 8, ErrorType.DATA_LOCK, "value not a incr"); seq = ArrayKits.bytesToLong(value) + step; } putDB(key_b, ArrayKits.longToBytes(seq), SstColumnFamily.DEFAULT); int time = (int) (System.currentTimeMillis() / 1000 + ttl); putDB(ArrayKits.addAll(HEAD_TTL, keyb), ArrayKits.intToBytes(time), SstColumnFamily.DEFAULT); setTimer(KeyEnum.KV_TIMER, time, key_b); commit(); checkTxCommit(); return seq; } finally { lock.unlock(lockEntity); release(); } } catch (Exception e) { checkTxRollBack(); throw e; } } }
RKv extends RBase { public long incr(String key, int step, int ttl) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { byte[] keyb = getKey(key); LockEntity lockEntity = lock.lock(key); try { start(); byte[] key_b = ArrayKits.addAll(HEAD_B, keyb); byte[] value = get(key); long seq; if (value == null) { seq = step; } else { DAssert.isTrue(value.length == 8, ErrorType.DATA_LOCK, "value not a incr"); seq = ArrayKits.bytesToLong(value) + step; } putDB(key_b, ArrayKits.longToBytes(seq), SstColumnFamily.DEFAULT); int time = (int) (System.currentTimeMillis() / 1000 + ttl); putDB(ArrayKits.addAll(HEAD_TTL, keyb), ArrayKits.intToBytes(time), SstColumnFamily.DEFAULT); setTimer(KeyEnum.KV_TIMER, time, key_b); commit(); checkTxCommit(); return seq; } finally { lock.unlock(lockEntity); release(); } } catch (Exception e) { checkTxRollBack(); throw e; } } protected  RKv(DB db); }
RKv extends RBase { public long incr(String key, int step, int ttl) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { byte[] keyb = getKey(key); LockEntity lockEntity = lock.lock(key); try { start(); byte[] key_b = ArrayKits.addAll(HEAD_B, keyb); byte[] value = get(key); long seq; if (value == null) { seq = step; } else { DAssert.isTrue(value.length == 8, ErrorType.DATA_LOCK, "value not a incr"); seq = ArrayKits.bytesToLong(value) + step; } putDB(key_b, ArrayKits.longToBytes(seq), SstColumnFamily.DEFAULT); int time = (int) (System.currentTimeMillis() / 1000 + ttl); putDB(ArrayKits.addAll(HEAD_TTL, keyb), ArrayKits.intToBytes(time), SstColumnFamily.DEFAULT); setTimer(KeyEnum.KV_TIMER, time, key_b); commit(); checkTxCommit(); return seq; } finally { lock.unlock(lockEntity); release(); } } catch (Exception e) { checkTxRollBack(); throw e; } } protected  RKv(DB db); void set(String key, byte[] value); long incr(String key, int step, int ttl); long incr(String key, int step); void set(Map<String, byte[]> map); void set(Map<String, byte[]> map, int ttl); void set(String key, byte[] value, int ttl); void ttl(String key, int ttl); Map<String, byte[]> get(List<String> keys); byte[] get(String key); byte[] getNoTTL(String key); void del(String key); void delPrefix(String key_); List<String> keys(String key_, int start, int limit); }
RKv extends RBase { public long incr(String key, int step, int ttl) throws KitDBException { checkTxStart(); try (CloseLock ignored = checkClose()) { byte[] keyb = getKey(key); LockEntity lockEntity = lock.lock(key); try { start(); byte[] key_b = ArrayKits.addAll(HEAD_B, keyb); byte[] value = get(key); long seq; if (value == null) { seq = step; } else { DAssert.isTrue(value.length == 8, ErrorType.DATA_LOCK, "value not a incr"); seq = ArrayKits.bytesToLong(value) + step; } putDB(key_b, ArrayKits.longToBytes(seq), SstColumnFamily.DEFAULT); int time = (int) (System.currentTimeMillis() / 1000 + ttl); putDB(ArrayKits.addAll(HEAD_TTL, keyb), ArrayKits.intToBytes(time), SstColumnFamily.DEFAULT); setTimer(KeyEnum.KV_TIMER, time, key_b); commit(); checkTxCommit(); return seq; } finally { lock.unlock(lockEntity); release(); } } catch (Exception e) { checkTxRollBack(); throw e; } } protected  RKv(DB db); void set(String key, byte[] value); long incr(String key, int step, int ttl); long incr(String key, int step); void set(Map<String, byte[]> map); void set(Map<String, byte[]> map, int ttl); void set(String key, byte[] value, int ttl); void ttl(String key, int ttl); Map<String, byte[]> get(List<String> keys); byte[] get(String key); byte[] getNoTTL(String key); void del(String key); void delPrefix(String key_); List<String> keys(String key_, int start, int limit); final static String HEAD; final static byte[] HEAD_TTL; final static byte[] HEAD_B; }